#4926 Multi Chat/LLM Sessions to Improve Upon Design Work and Problem Solving
#4926 Multi Chat/LLM Sessions to Improve Upon Design Work and Problem Solving
- How can I use multiple chat sessions across one or more LLMs to improve on a design and/or problem solving session?
- Write an executable Python workflow orchestrating a multi-model debate between a generator, critic, and synthesizer.
- Add automated execution logging to this Python script that writes a detailed Markdown audit trail with diffs between design iterations.
- Would this work for two problem statements, 1) Improve the aerodynamics of a bicycle while maintaining rider pedaling efficiency and comfort, and 2) Increase the energy density of an electric vehicle battery system while not increasing system weight and cost ?
- Provide a prompt to dynamically generate the pydantic class needed for a specified problem
- Modify MultiModelDebateEngine to accept arbitrary problem statements, dynamically generate and compile domain schemas, and execute the debate using those generated models.
- Update DynamicMultiModelDebateEngine with an automatic repair loop that feeds AST validation or compilation errors back to the metaprogrammer for self-correction.
- Integrate SymPy equation checking into the schema validation loop so physical laws (like aerodynamic power equations) are formally verified before the debate begins.
#4926 How can I use multiple chat sessions across one or more LLMs to improve on a design and/or problem solving session?
Using multiple chat sessions—whether running parallel sessions within the same model or cross-examining across fundamentally different model families (e.g., Claude, Gemini, GPT, open-weights)—breaks the cognitive lock-in, agreeable sycophancy, and context contamination that degrade long, single-thread problem-solving sessions.
Core Architectures for Multi-Session Workflows
1. Adversarial Peer Review (Generator–Auditor Loop)
A single chat session tends to justify its own prior assumptions. Separating generation from verification across distinct contexts forces objective scrutiny.
Session A (Generator / Designer): Prompted with domain constraints, functional objectives, and requirements. Task: Generate candidate designs or algorithms.
Session B (Auditor / Red Team): Operates completely blank-slate, without seeing Session A's reasoning steps. Provide only the finished design from Session A alongside a hostile evaluation prompt:
"Analyze this design strictly for failure modes, unstated assumptions, race conditions, edge-case bottlenecks, and physical/computational infeasibility. Do not offer compliments; list verified defects and rank them by severity."
Cycle: Take Session B's prioritized defects back to Session A to engineer counter-measures, repeating until Session B reports zero critical flaws.
2. Blind Divergent Council (Parallel Ideation & Matrix Scoring)
Running a problem statement in one session limits your solution space to that session’s initial probabilistic path.
Sessions 1, 2, and 3 (Independent Generators): Run identical problem criteria in 3 separate instances (ideally differing model families or system prompts emphasizing different methodologies, such as first-principles decomposition vs. analogical transfer vs. conservative modular design). Instruct each to produce 2 distinct, uncompromised concepts.
Session 4 (The Judge / Synthesizer): Paste all 6 concepts into a fresh session with an explicit evaluation rubric (e.g., implementation complexity, latency, maintainability, failure recovery). Task the model with constructing a comparative trade study matrix and identifying complementary elements that can be hybridized into a dominant design.
3. Recursive Decomposition Pipeline (Hierarchical Chains)
Avoid asking a single session to handle both high-level system architecture and low-level component execution. Context windows fill, instruction-following slips, and architectural trade-offs get blurred with syntax details.
Session A (Architectural Decomposition): Deconstructs the system into decoupled modules, establishing exact input/output contracts, state invariants, and boundary schemas.
Sessions B, C, D (Specialized Component Solvers): Each session is given only the global interface contract and its assigned sub-module requirements. This keeps the attention mechanism fully focused on solving that specific component's internal logic.
Session E (Integration Verification): Takes the components developed in B, C, and D and simulates edge-case interaction and data flows across the boundaries defined by Session A.
Execution Protocols
| Challenge | Failure in Single Chat | Multi-Session Mitigation |
| Sycophancy / Confirmation Bias | The model agrees with your leading feedback or defends its initial flaws. | Use a fresh session with an Adversarial Auditor role that has zero commitment to the prior output. |
| Context Drift & Degraded Attention | After 10+ turns, earlier constraints are forgotten or diluted. | Use Summary Checkpoints: compress settled decisions into a formal specification doc and seed a clean session with it. |
| Blind Spots Across Model Families | Models with similar post-training share inductive biases. | Cross-pollinate: use a reasoning-heavy frontier model for proof/architecture verification and an alternative family for creative trade-offs or edge-case enumeration. |
Practical Operating Rules
Pass State via Formal Artifacts, Not Chat Logs: When porting ideas between sessions, never paste raw conversational back-and-forth. Extract only structured artifacts—JSON specifications, Pydantic schemas, state-transition tables, or mathematical formulation blocks. This strips out conversational fluff and prevents downstream sessions from anchoring to conversational nuances.
Strip Attribution When Cross-Checking: When giving Session B an output from Session A, do not say "Another AI suggested this." Present it neutrally: "Evaluate the following technical proposal against [X constraints]." This avoids triggering conversational deference or generic meta-commentary about AI capabilities.
Enforce Deterministic Rubrics: Never ask an evaluator session "Is this good?" Instruct it to run explicit checklist validations:
Verification against mandatory requirements (Pass/Fail).
Worst-case environmental/operational stress.
Modularity and interface coupling.
Write an executable Python workflow orchestrating a multi-model debate between a generator, critic, and synthesizer.
An executable orchestration script using the google-genai SDK and Pydantic v2 schemas. It implements a closed-loop deliberation pipeline where:
Generator drafts an initial architecture proposal matching functional constraints.
Adversarial Critic inspects the proposal in a context-isolated call, generating structured failure modes, edge-case bottlenecks, and a veto status.
Loop Governor routes critical defects back to the Generator for targeted revisions (up to
max_iterations).Synthesizer reconciles the design history into a finalized technical specification with an integrated risk audit.
import os
import json
from typing import List, Optional
from pydantic import BaseModel, Field
from google import genai
from google.genai import types
# ---------------------------------------------------------------------------
# Structured Schemas (Pydantic v2)
# ---------------------------------------------------------------------------
class InterfaceSpec(BaseModel):
name: str = Field(description="Name of the interface or endpoint")
inputs: List[str] = Field(description="Input parameter types/contracts")
outputs: List[str] = Field(description="Return values or emitted state")
class ComponentSpec(BaseModel):
name: str = Field(description="Component name")
responsibility: str = Field(description="Core functional responsibility")
dependencies: List[str] = Field(description="Direct upstream/downstream dependencies")
class ProposalSchema(BaseModel):
design_title: str
architectural_summary: str
components: List[ComponentSpec]
interfaces: List[InterfaceSpec]
state_invariants: List[str] = Field(description="Invariants that must hold across executions")
class Defect(BaseModel):
category: str = Field(description="e.g., Race Condition, Single Point of Failure, Boundary Case")
severity: str = Field(description="CRITICAL, HIGH, MEDIUM, or LOW")
description: str = Field(description="Precise technical mechanism of the defect")
recommended_mitigation: str = Field(description="Concrete architectural countermeasure")
class AuditReportSchema(BaseModel):
has_critical_defects: bool = Field(description="True if any CRITICAL or HIGH severity flaws exist")
identified_defects: List[Defect]
critique_summary: str
class FinalSpecificationSchema(BaseModel):
title: str
executive_summary: str
reconciled_components: List[ComponentSpec]
reconciled_interfaces: List[InterfaceSpec]
guaranteed_invariants: List[str]
residual_risks: List[str] = Field(description="Known operational limitations accepted in the final design")
# ---------------------------------------------------------------------------
# Orchestration Engine
# ---------------------------------------------------------------------------
class MultiModelDebateEngine:
def __init__(self, api_key: Optional[str] = None):
self.client = genai.Client(api_key=api_key or os.environ.get("GEMINI_API_KEY"))
# Using frontier reasoning models for generation/synthesis, standard for auditor
self.generator_model = "gemini-2.5-pro"
self.critic_model = "gemini-2.5-flash"
self.synthesizer_model = "gemini-2.5-pro"
def _call_structured_llm(self, model: str, system_instruction: str, prompt: str, schema: type[BaseModel]):
"""Executes an isolated, structured call with zero cross-session context leaks."""
response = self.client.models.generate_content(
model=model,
contents=prompt,
config=types.GenerateContentConfig(
system_instruction=system_instruction,
response_mime_type="application/json",
response_schema=schema,
temperature=0.2,
),
)
return schema.model_validate_json(response.text)
def generate_initial(self, requirements: str) -> ProposalSchema:
sys_inst = (
"You are a Principal Systems Architect. Produce a rigorous, decoupled, "
"and robust architecture based strictly on the provided functional constraints. "
"Do not output conversational text; strictly adhere to the output schema."
)
prompt = f"Functional Requirements:\n{requirements}\n\nFormulate a complete architectural proposal."
return self._call_structured_llm(self.generator_model, sys_inst, prompt, ProposalSchema)
def audit_proposal(self, requirements: str, proposal: ProposalSchema) -> AuditReportSchema:
sys_inst = (
"You are a Hostile Quality Auditor and Systems Red-Teamer. Analyze the proposed design "
"strictly for unstated assumptions, failure modes, data consistency races, bottleneck points, "
"and unhandled boundary states against the requirements. Do not offer praise. "
"Mark has_critical_defects=True if any flaw compromises stability, correctness, or availability."
)
prompt = (
f"Original Requirements:\n{requirements}\n\n"
f"Candidate Architecture Proposal:\n{proposal.model_dump_json(indent=2)}\n\n"
f"Execute an adversarial audit."
)
return self._call_structured_llm(self.critic_model, sys_inst, prompt, AuditReportSchema)
def revise_proposal(self, requirements: str, previous_proposal: ProposalSchema, audit: AuditReportSchema) -> ProposalSchema:
sys_inst = (
"You are a Principal Systems Architect addressing hostile audit findings. "
"Directly remediate all identified defects by restructuring interfaces, components, "
"or state invariants while keeping the design decoupled and compliant."
)
defects_text = "\n".join(
[f"- [{d.severity}] {d.category}: {d.description} -> Fix: {d.recommended_mitigation}"
for d in audit.identified_defects]
)
prompt = (
f"Functional Requirements:\n{requirements}\n\n"
f"Prior Architecture:\n{previous_proposal.model_dump_json(indent=2)}\n\n"
f"Auditor Defects to Resolve:\n{defects_text}\n\n"
f"Generate a revised architectural proposal resolving these issues."
)
return self._call_structured_llm(self.generator_model, sys_inst, prompt, ProposalSchema)
def synthesize_final(self, requirements: str, proposal: ProposalSchema, audit_history: List[AuditReportSchema]) -> FinalSpecificationSchema:
sys_inst = (
"You are an Executive Technical Lead. Synthesize the vetted design iteration into "
"a production-ready technical specification. Reconcile remaining minor flaws into "
"explicit residual risks."
)
history_summary = []
for i, audit in enumerate(audit_history, 1):
history_summary.append(f"Audit Round {i}: {audit.critique_summary}")
prompt = (
f"Requirements:\n{requirements}\n\n"
f"Final Vetted Architecture:\n{proposal.model_dump_json(indent=2)}\n\n"
f"Audit Iteration History:\n" + "\n".join(history_summary) + "\n\n"
f"Produce the final reconciled technical specification."
)
return self._call_structured_llm(self.synthesizer_model, sys_inst, prompt, FinalSpecificationSchema)
def run_debate_pipeline(self, requirements: str, max_iterations: int = 2) -> FinalSpecificationSchema:
print("[1/4] Generating initial candidate proposal...")
proposal = self.generate_initial(requirements)
print(f" Initial Design: {proposal.design_title}")
audit_history: List[AuditReportSchema] = []
for round_idx in range(1, max_iterations + 1):
print(f"\n[2/4] Executing adversarial audit round {round_idx} of {max_iterations}...")
audit = self.audit_proposal(requirements, proposal)
audit_history.append(audit)
critical_count = sum(1 for d in audit.identified_defects if d.severity in ("CRITICAL", "HIGH"))
print(f" Defects Identified: {len(audit.identified_defects)} total ({critical_count} critical/high)")
if not audit.has_critical_defects or round_idx == max_iterations:
if not audit.has_critical_defects:
print(" Passed audit: Zero critical architectural defects remain.")
else:
print(" Max iterations reached. Proceeding to synthesis with documented residual risks.")
break
print(f"[3/4] Rerouting defects to Generator for iteration {round_idx + 1}...")
proposal = self.revise_proposal(requirements, proposal, audit)
print("\n[4/4] Synthesizing final technical specification...")
final_spec = self.synthesizer_final(requirements, proposal, audit_history)
print("Pipeline complete.\n")
return final_spec
# ---------------------------------------------------------------------------
# Execution Entry Point
# ---------------------------------------------------------------------------
if __name__ == "__main__":
test_requirements = (
"Design a distributed task-scheduling engine that coordinates long-running async jobs "
"across a heterogeneous worker pool. Requirements:\n"
"- Zero duplicate job executions under worker node network partitions (split-brain).\n"
"- Sub-second recovery upon worker death.\n"
"- At-least-once status telemetry dispatch to clients via WebSocket.\n"
"- Storage layer must survive unexpected power loss without state corruption."
)
# Requires GEMINI_API_KEY environment variable
orchestrator = MultiModelDebateEngine()
final_spec = orchestrator.run_debate_pipeline(test_requirements, max_iterations=2)
# Output formatted specification
print(f"=== {final_spec.title} ===")
print(f"\nExecutive Summary:\n{final_spec.executive_summary}\n")
print("Guaranteed Invariants:")
for inv in final_spec.guaranteed_invariants:
print(f" * {inv}")
print("\nResidual Risks:")
for risk in final_spec.residual_risks:
print(f" ! {risk}")
Core Architectural Features
Context Isolation: The Auditor never sees the Generator's internal chain-of-thought or conversational prompts. It evaluates purely against raw inputs/outputs and functional requirements, preventing bias transfer.
Deterministic Contract Enforcement: Every stage leverages Pydantic schemas via Gemini's native
response_schemaparameter to guarantee that the loop state is passed as strongly-typed, verifiable JSON instead of conversational prose.Governor Termination Conditions: The debate terminates automatically if
has_critical_defectsis resolved asFalse, or cleanly drops through to risk reconciliation oncemax_iterationsis hit.
Add automated execution logging to this Python script that writes a detailed Markdown audit trail with diffs between design iterations.
Here is the updated, standalone script. It introduces a dedicated AuditLogger that captures all prompts, raw schema payloads, structured evaluation results, and computes line-by-line unified diffs between successive design proposals using Python's standard difflib.
import os
import json
import difflib
from datetime import datetime
from typing import List, Optional
from pydantic import BaseModel, Field
from google import genai
from google.genai import types
# ---------------------------------------------------------------------------
# Structured Schemas (Pydantic v2)
# ---------------------------------------------------------------------------
class InterfaceSpec(BaseModel):
name: str = Field(description="Name of the interface or endpoint")
inputs: List[str] = Field(description="Input parameter types/contracts")
outputs: List[str] = Field(description="Return values or emitted state")
class ComponentSpec(BaseModel):
name: str = Field(description="Component name")
responsibility: str = Field(description="Core functional responsibility")
dependencies: List[str] = Field(description="Direct upstream/downstream dependencies")
class ProposalSchema(BaseModel):
design_title: str
architectural_summary: str
components: List[ComponentSpec]
interfaces: List[InterfaceSpec]
state_invariants: List[str] = Field(description="Invariants that must hold across executions")
class Defect(BaseModel):
category: str = Field(description="e.g., Race Condition, Single Point of Failure, Boundary Case")
severity: str = Field(description="CRITICAL, HIGH, MEDIUM, or LOW")
description: str = Field(description="Precise technical mechanism of the defect")
recommended_mitigation: str = Field(description="Concrete architectural countermeasure")
class AuditReportSchema(BaseModel):
has_critical_defects: bool = Field(description="True if any CRITICAL or HIGH severity flaws exist")
identified_defects: List[Defect]
critique_summary: str
class FinalSpecificationSchema(BaseModel):
title: str
executive_summary: str
reconciled_components: List[ComponentSpec]
reconciled_interfaces: List[InterfaceSpec]
guaranteed_invariants: List[str]
residual_risks: List[str] = Field(description="Known operational limitations accepted in the final design")
# ---------------------------------------------------------------------------
# Markdown Audit Logger with Diff Tracking
# ---------------------------------------------------------------------------
class AuditLogger:
def __init__(self, output_path: str = "debate_audit_report.md"):
self.output_path = output_path
self.entries: List[str] = []
self._init_log()
def _init_log(self):
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
self.entries.append(f"# Multi-Model Architectural Debate & Audit Trail\n")
self.entries.append(f"**Execution Timestamp:** `{timestamp}` \n")
self.entries.append(f"---\n")
def log_requirements(self, requirements: str):
self.entries.append("## 1. System Requirements & Invariants\n")
self.entries.append("```text")
self.entries.append(requirements.strip())
self.entries.append("```\n")
def log_initial_proposal(self, proposal: ProposalSchema):
self.entries.append("## 2. Iteration 0: Initial Candidate Design\n")
self.entries.append(f"### Design: {proposal.design_title}\n")
self.entries.append(f"**Summary:** {proposal.architectural_summary}\n")
self.entries.append("```json")
self.entries.append(proposal.model_dump_json(indent=2))
self.entries.append("```\n")
def log_audit_round(self, round_num: int, audit: AuditReportSchema):
self.entries.append(f"### Audit Round {round_num}: Adversarial Red-Team Review\n")
status = "**REJECTED (Critical Defects Found)**" if audit.has_critical_defects else "**ACCEPTED (Clean)**"
self.entries.append(f"**Verdict:** {status} \n")
self.entries.append(f"**Summary:** {audit.critique_summary}\n")
if audit.identified_defects:
self.entries.append("| Severity | Category | Description | Recommended Mitigation |")
self.entries.append("|:---|:---|:---|:---|")
for d in audit.identified_defects:
self.entries.append(f"| `{d.severity}` | {d.category} | {d.description} | {d.recommended_mitigation} |")
self.entries.append("\n")
def log_iteration_diff(self, round_num: int, prev_proposal: ProposalSchema, new_proposal: ProposalSchema):
self.entries.append(f"## Iteration {round_num}: Architectural Revision & Diff\n")
self.entries.append(f"### Revised Design: {new_proposal.design_title}\n")
self.entries.append(f"**Updated Summary:** {new_proposal.architectural_summary}\n")
prev_lines = prev_proposal.model_dump_json(indent=2).splitlines(keepends=True)
new_lines = new_proposal.model_dump_json(indent=2).splitlines(keepends=True)
diff = list(difflib.unified_diff(
prev_lines,
new_lines,
fromfile=f"Iteration_{round_num-1}.json",
tofile=f"Iteration_{round_num}.json",
lineterm=""
))
self.entries.append("#### Structural State Diff (`diff`)\n")
self.entries.append("```diff")
if diff:
self.entries.extend([line.rstrip() for line in diff])
else:
self.entries.append("No structural differences detected.")
self.entries.append("```\n")
def log_final_specification(self, final_spec: FinalSpecificationSchema):
self.entries.append("## 3. Final Reconciled Technical Specification\n")
self.entries.append(f"### {final_spec.title}\n")
self.entries.append(f"{final_spec.executive_summary}\n")
self.entries.append("#### Guaranteed Invariants\n")
for inv in final_spec.guaranteed_invariants:
self.entries.append(f"- {inv}")
self.entries.append("\n")
self.entries.append("#### Residual Risks & Accepted Trade-offs\n")
for risk in final_spec.residual_risks:
self.entries.append(f"- ⚠️ {risk}")
self.entries.append("\n")
self.entries.append("#### Full Production Spec Payload\n")
self.entries.append("```json")
self.entries.append(final_spec.model_dump_json(indent=2))
self.entries.append("```\n")
def write_to_disk(self):
with open(self.output_path, "w", encoding="utf-8") as f:
f.write("\n".join(self.entries))
print(f"Audit log saved: {os.path.abspath(self.output_path)}")
# ---------------------------------------------------------------------------
# Orchestration Engine
# ---------------------------------------------------------------------------
class MultiModelDebateEngine:
def __init__(self, api_key: Optional[str] = None, log_file: str = "debate_audit_report.md"):
self.client = genai.Client(api_key=api_key or os.environ.get("GEMINI_API_KEY"))
self.generator_model = "gemini-2.5-pro"
self.critic_model = "gemini-2.5-flash"
self.synthesizer_model = "gemini-2.5-pro"
self.logger = AuditLogger(output_path=log_file)
def _call_structured_llm(self, model: str, system_instruction: str, prompt: str, schema: type[BaseModel]):
response = self.client.models.generate_content(
model=model,
contents=prompt,
config=types.GenerateContentConfig(
system_instruction=system_instruction,
response_mime_type="application/json",
response_schema=schema,
temperature=0.2,
),
)
return schema.model_validate_json(response.text)
def generate_initial(self, requirements: str) -> ProposalSchema:
sys_inst = (
"You are a Principal Systems Architect. Produce a rigorous, decoupled, "
"and robust architecture based strictly on the provided functional constraints. "
"Do not output conversational text; strictly adhere to the output schema."
)
prompt = f"Functional Requirements:\n{requirements}\n\nFormulate a complete architectural proposal."
return self._call_structured_llm(self.generator_model, sys_inst, prompt, ProposalSchema)
def audit_proposal(self, requirements: str, proposal: ProposalSchema) -> AuditReportSchema:
sys_inst = (
"You are a Hostile Quality Auditor and Systems Red-Teamer. Analyze the proposed design "
"strictly for unstated assumptions, failure modes, data consistency races, bottleneck points, "
"and unhandled boundary states against the requirements. Do not offer praise. "
"Mark has_critical_defects=True if any flaw compromises stability, correctness, or availability."
)
prompt = (
f"Original Requirements:\n{requirements}\n\n"
f"Candidate Architecture Proposal:\n{proposal.model_dump_json(indent=2)}\n\n"
f"Execute an adversarial audit."
)
return self._call_structured_llm(self.critic_model, sys_inst, prompt, AuditReportSchema)
def revise_proposal(self, requirements: str, previous_proposal: ProposalSchema, audit: AuditReportSchema) -> ProposalSchema:
sys_inst = (
"You are a Principal Systems Architect addressing hostile audit findings. "
"Directly remediate all identified defects by restructuring interfaces, components, "
"or state invariants while keeping the design decoupled and compliant."
)
defects_text = "\n".join(
[f"- [{d.severity}] {d.category}: {d.description} -> Fix: {d.recommended_mitigation}"
for d in audit.identified_defects]
)
prompt = (
f"Functional Requirements:\n{requirements}\n\n"
f"Prior Architecture:\n{previous_proposal.model_dump_json(indent=2)}\n\n"
f"Auditor Defects to Resolve:\n{defects_text}\n\n"
f"Generate a revised architectural proposal resolving these issues."
)
return self._call_structured_llm(self.generator_model, sys_inst, prompt, ProposalSchema)
def synthesize_final(self, requirements: str, proposal: ProposalSchema, audit_history: List[AuditReportSchema]) -> FinalSpecificationSchema:
sys_inst = (
"You are an Executive Technical Lead. Synthesize the vetted design iteration into "
"a production-ready technical specification. Reconcile remaining minor flaws into "
"explicit residual risks."
)
history_summary = [f"Audit Round {i}: {a.critique_summary}" for i, a in enumerate(audit_history, 1)]
prompt = (
f"Requirements:\n{requirements}\n\n"
f"Final Vetted Architecture:\n{proposal.model_dump_json(indent=2)}\n\n"
f"Audit Iteration History:\n" + "\n".join(history_summary) + "\n\n"
f"Produce the final reconciled technical specification."
)
return self._call_structured_llm(self.synthesizer_model, sys_inst, prompt, FinalSpecificationSchema)
def run_debate_pipeline(self, requirements: str, max_iterations: int = 2) -> FinalSpecificationSchema:
self.logger.log_requirements(requirements)
print("[1/4] Generating initial candidate proposal...")
proposal = self.generate_initial(requirements)
self.logger.log_initial_proposal(proposal)
print(f" Initial Design: {proposal.design_title}")
audit_history: List[AuditReportSchema] = []
for round_idx in range(1, max_iterations + 1):
print(f"\n[2/4] Executing adversarial audit round {round_idx} of {max_iterations}...")
audit = self.audit_proposal(requirements, proposal)
audit_history.append(audit)
self.logger.log_audit_round(round_idx, audit)
critical_count = sum(1 for d in audit.identified_defects if d.severity in ("CRITICAL", "HIGH"))
print(f" Defects: {len(audit.identified_defects)} total ({critical_count} critical/high)")
if not audit.has_critical_defects or round_idx == max_iterations:
if not audit.has_critical_defects:
print(" Audit Passed: No critical architectural defects remaining.")
else:
print(" Max iterations reached. Proceeding to final synthesis.")
break
print(f"[3/4] Rerouting defects to Generator for revision {round_idx}...")
revised_proposal = self.revise_proposal(requirements, proposal, audit)
self.logger.log_iteration_diff(round_idx, proposal, revised_proposal)
proposal = revised_proposal
print("\n[4/4] Synthesizing final technical specification...")
final_spec = self.synthesizer_final(requirements, proposal, audit_history)
self.logger.log_final_specification(final_spec)
self.logger.write_to_disk()
print("Pipeline complete.\n")
return final_spec
# ---------------------------------------------------------------------------
# Execution Entry Point
# ---------------------------------------------------------------------------
if __name__ == "__main__":
test_requirements = (
"Design a distributed task-scheduling engine coordinating long-running async jobs "
"across a heterogeneous worker pool. Requirements:\n"
"- Zero duplicate job executions under worker node network partitions (split-brain).\n"
"- Sub-second recovery upon worker death.\n"
"- At-least-once status telemetry dispatch to clients via WebSocket.\n"
"- Storage layer must survive unexpected power loss without state corruption."
)
orchestrator = MultiModelDebateEngine(log_file="distributed_scheduler_audit.md")
spec = orchestrator.run_debate_pipeline(test_requirements, max_iterations=2)Would this work for two problem statements, 1) Improve the aerodynamics of a bicycle while maintaining rider pedaling efficiency and comfort, and 2) Increase the energy density of an electric vehicle battery system while not increasing system weight and cost ?
Yes, the architecture works for both problem statements, but each exposes distinct engineering contradictions that require tweaking how the Pydantic schemas and Auditor prompts are structured.
In pure software architectures, state invariants are logical contracts (e.g., "no duplicate executions"). For physical and electro-chemical domains, state invariants become governing physical laws and coupled trade-offs. If left unconstrained, a generic generator will produce hand-waving "free lunches" (e.g., claiming a super-low aerodynamic posture without biomechanical power loss, or quoting cell-level silicon-anode density gains without accounting for thermal/structural pack overhead).
Here is an analysis of how the debate loop operates on each problem, along with the concrete schema adjustments needed to make the auditor hostile to unrealistic physics.
Problem 1: Bicycle Aerodynamics vs. Biomechanical Efficiency & Comfort
The Core Contradiction
Aerodynamic drag (F_d = 1/2 rho v^2 C_d A) demands minimizing frontal surface area (A) and streamlining airflow over the torso and head. However, closing the hip angle (theta_hip) below roughly 45-degrees compresses the diaphragm, restricts peak lung tidal volume, and impedes hip-extensor muscle recruitment (gluteus maximus and hamstrings), reducing sustainable functional threshold power (FTP). Furthermore, aggressive cervical spine hyperextension causes neck/shoulder fatigue, destroying long-duration compliance.
How the Pipeline Behaves
| Stage | Expected Generator Output | Auditor Red-Team Focus | Necessary Mitigation in Revision |
| Initial Proposal | Extreme dropped cockpit, narrowed aero bars, integrated cockpit fairings, high saddle-to-bar drop (>12 cm). | Flags hip angle impingement, loss of pelvic rotation, reduced respiratory capacity, and unviable neck extension angles. | Shortened crank arms (165 mm or 160 mm) to open hip angle at top dead center; forward saddle offset (>78-degrees effective seat tube angle) to rotate torso forward while preserving open hip kinematics. |
| Iteration 1 | Introduces steep effective seat angle and narrow cockpit with forearm cups. | Flags lateral handling instability in crosswinds (high yaw angle shedding), restricted pectoral expansion, and road vibration transmission fatigue. | Adds tuned compliance zones in the seatpost/stays, flared extensions, and aerodynamically shrouded storage to smooth airflow over the bottom bracket. |
| Final Synthesis | Reconciled balance: C_d A reduction achieved through posture rotation rather than acute joint compression; vibration damping specs defined. |
Necessary Prompt / Schema Adjustment
Replace generic software fields with explicit biomechanical and fluid dynamics metrics:
class BiomechanicalInvariant(BaseModel):
min_hip_angle_at_tdc_deg: float = Field(ge=45.0, description="Minimum hip angle at Top Dead Center to prevent power loss")
max_cervical_extension_deg: float = Field(le=35.0, description="Max neck angle for sustained comfort")
handling_crosswind_yaw_limit_deg: float = Field(description="Stable yaw handling range")
Problem 2: EV Battery Pack Energy Density vs. Total Weight & Cost
The Core Contradiction
Gravimetric energy density (Wh/kg) and volumetric density (Wh/L) can be increased at the cell chemistry level (e.g., high-nickel NMC 811, silicon-graphite composite anodes, or semi-solid electrolytes). However:
Silicon anodes swell (>100--}300% volumetric expansion during lithiation), requiring rigid mechanical pack containment or high compressive foam padding, which adds structural parasitic weight.
High-nickel cathodes and dense packing increase exothermic runaway risks, demanding heavier cooling plates, redundant flame barriers, and expensive aerogel insulation.
Advanced chemistries and laser-welded cell-to-pack (CTP) designs introduce manufacturing yield fallout and exotic raw material costs ($/kWh).
How the Pipeline Behaves
| Stage | Expected Generator Output | Auditor Red-Team Focus | Necessary Mitigation in Revision |
| Initial Proposal | Transition to high-silicon anode cells, Cell-to-Pack (CTP) or Cell-to-Chassis (CTC) architecture eliminating modular sub-enclosures. | Flags cycle-life degradation from anode expansion, inadequate thermal isolation during single-cell thermal runaway, and ballooning material costs. | Integrates micro-channel cold plates directly into structural cross-members; introduces breathing foam/elastomeric buffers; balances silicon content to <15 wt%. |
| Iteration 1 | Adds structural thermal management and elastomeric gap pads. | Flags pack-level mass creep offsetting cell-level gravimetric gains, serviceability failure (non-repairable bonded adhesive), and high tooling capex. | Switches to high-strength thin-gauge stamped aluminum or composite top cover; standardizes block-level disconnects; leverages low-cobalt/manganese-rich cathode blends to preserve pack-level cost parity. |
| Final Synthesis | Reconciled net energy density gain at the system level (accounting for structural and thermal overhead) while holding $/kWh flat. |
Necessary Prompt / Schema Adjustment
Ensure the auditor forces a distinction between cell-level and pack-level metrics:
class BatterySystemMetrics(BaseModel):
cell_gravimetric_density_wh_kg: float
pack_gravimetric_density_wh_kg: float
pack_to_cell_mass_ratio: float = Field(description="Must realistically account for thermal, enclosure, and BMS mass")
cost_per_kwh_pack_usd: float = Field(description="Must not exceed baseline benchmark")
thermal_runaway_containment_strategy: str
Summary of System Performance on Physical Systems
The debate workflow excels on these problems precisely because single LLM chats frequently commit optimistic cherry-picking—they recommend extreme aerodynamic positions or high-energy chemistries while ignoring the downstream physical penalties.
By deploying an adversarial auditor prompted specifically to police parasitic overhead, biomechanical fatigue, and second-order physical consequences, the engine forces the generator to address root-cause engineering contradictions rather than offering superficial compromises.
Provide a prompt to dynamically generate the pydantic class needed for a specified problem
Here is a prompt designed to act as a Metaprogramming Schema Engineer. It analyzes any engineering contradiction or domain problem and synthesizes self-contained, validated Pydantic v2 classes tailored for that problem's specific trade study, physical invariants, and failure modes.
Dynamic Schema Generation Meta-Prompt
You are an expert Systems Metaprogrammer and Computational Ontologist specializing in Pydantic v2 and domain-specific engineering data models.
### OBJECTIVE
Given a target problem statement and its governing engineering contradictions, generate a production-ready, self-contained Python module defining the complete set of Pydantic v2 schemas required to run a closed-loop multi-agent debate (Generator -> Auditor -> Synthesizer).
### INPUT SPECIFICATION
Problem Statement:
"""
{INSERT_PROBLEM_STATEMENT_HERE}
"""
Governing Trade-Offs / Coupled Physical Constraints:
"""
{INSERT_CONSTRAINTS_AND_TRADEOFFS_HERE}
"""
### SCHEMA DESIGN DIRECTIVES
1. Tailor Strictly to the Problem Domain:
- Do NOT use generic software fields (e.g., avoid vague labels like "feature_1", "component_a", or generic "string_data").
- Extract domain-specific quantitative parameters, physical state variables, efficiency trade-offs, and failure mode classes directly reflecting the governing physics, materials, and biomechanical/mechanical bounds.
2. Enforce Formal Pydantic v2 Validation:
- Use standard Python type hints (`float`, `int`, `str`, `List`, `Dict`, `Optional`).
- Use `Field(..., description="...", ge=..., le=...)` with realistic physical units and boundary ranges to trap unphysical "free lunch" proposals (e.g., angles, efficiencies between 0.0 and 1.0, non-negative temperatures, parasitic mass scaling).
- Apply `@field_validator` or `@model_validator(mode="after")` to enforce coupled trade-offs (e.g., if density increases, verify structural/thermal mass scaling or validate that efficiency does not violate the second law of thermodynamics).
3. Required Model Architecture:
- `DomainParameters`: Quantitative specifications, physical dimensions, material properties, and operational thresholds with strict numerical constraints.
- `TradeOffInvariants`: Coupled mathematical invariants that must hold true (e.g., biomechanical angles, mass conservation, thermal equilibrium, pack-to-cell ratios).
- `DomainProposalSchema`: The candidate technical solution payload containing system architecture, subsystem allocations, operational metrics, and claimed invariant satisfaction.
- `DomainDefectSchema`: Detailed failure modes, categorized specifically for this domain (e.g., fatigue, thermal runaway, flow separation, ergonomic impingement), with severity ratings (`CRITICAL`, `HIGH`, `MEDIUM`, `LOW`) and actionable physical countermeasures.
- `DomainAuditReportSchema`: Auditor output schema containing `has_critical_defects: bool`, an itemized list of `DomainDefectSchema`, and an adversarial technical critique.
- `DomainFinalSpecificationSchema`: Reconciled production specification schema with final parameters, validated invariant proofs, and accepted residual operational risks.
4. Output Constraints:
- Return valid, executable Python code enclosed inside a single ```python ... ``` block.
- Include all necessary imports (`from pydantic import BaseModel, Field, field_validator, model_validator`, `from typing import List, Optional`, etc.).
- Ensure the generated code can be executed directly or parsed cleanly via dynamic loading (`exec()` or dynamic module imports) without undefined dependencies.
Python Dynamic Compiler Implementation
To run this prompt programmatically and compile the returned Pydantic classes at runtime for immediate use in the debate engine:
import re
from typing import Dict, Any, Type
from pydantic import BaseModel
from google import genai
from google.genai import types
def generate_dynamic_schemas(
problem_statement: str,
governing_tradeoffs: str,
client: genai.Client,
model_name: str = "gemini-2.5-pro"
) -> Dict[str, Type[BaseModel]]:
"""
Prompts the LLM to generate domain-specific Pydantic v2 code,
executes it in an isolated namespace, and returns the schema classes.
"""
meta_prompt = f"""
You are an expert Systems Metaprogrammer and Computational Ontologist specializing in Pydantic v2.
Generate a self-contained Python module defining Pydantic v2 schemas for an adversarial engineering debate on this problem:
Problem:
{problem_statement}
Trade-offs / Constraints:
{governing_tradeoffs}
Follow these rules:
1. Create domain-specific quantitative schemas:
- `DomainProposalSchema` (system breakdown, physical parameters, invariant bounds)
- `DomainDefectSchema` (domain-specific defect classifications and mitigations)
- `DomainAuditReportSchema` (contains has_critical_defects: bool, defects: List[DomainDefectSchema], critique_summary: str)
- `DomainFinalSpecificationSchema` (reconciled specs, invariants, residual risks)
2. Use pydantic.Field with explicit bounds (`ge`, `le`, `gt`) and physical units in descriptions.
3. Include all imports (`from pydantic import BaseModel, Field`, `from typing import List, Optional, Dict`).
4. Return ONLY valid Python code inside a ```python ``` block.
"""
response = client.models.generate_content(
model=model_name,
contents=meta_prompt,
config=types.GenerateContentConfig(
temperature=0.1,
)
)
# Extract python code block
match = re.search(r"```python\s*(.*?)\s*```", response.text, re.DOTALL)
code_str = match.group(1) if match else response.text
# Execute in a dedicated local namespace
namespace: Dict[str, Any] = {}
exec(code_str, namespace)
# Extract the generated Pydantic models
schemas = {
name: cls
for name, cls in namespace.items()
if isinstance(cls, type) and issubclass(cls, BaseModel) and cls is not BaseModel
}
return schemasModify MultiModelDebateEngine to accept arbitrary problem statements, dynamically generate and compile domain schemas, and execute the debate using those generated models.
Here is the complete, self-contained architecture. It integrates runtime schema synthesis, AST security validation, in-memory compilation via an isolated namespace, and dynamic payload handling using Pydantic v2's model_validate_json() and model_dump_json().
import os
import re
import ast
import json
import difflib
from datetime import datetime
from typing import List, Dict, Any, Type, Optional
from pydantic import BaseModel
from google import genai
from google.genai import types
# ---------------------------------------------------------------------------
# AST Security Validator
# ---------------------------------------------------------------------------
class SchemaSecurityVisitor(ast.NodeVisitor):
"""
Enforces that dynamically generated code contains strictly declarative Pydantic schemas,
blocking file I/O, subprocesses, network calls, eval/exec, and arbitrary system imports.
"""
ALLOWED_MODULES = {"pydantic", "typing", "typing_extensions", "math", "datetime", "enum"}
DISALLOWED_FUNCTIONS = {"exec", "eval", "compile", "open", "__import__", "getattr", "setattr"}
def visit_Import(self, node: ast.Import):
for alias in node.names:
base_module = alias.name.split(".")[0]
if base_module not in self.ALLOWED_MODULES:
raise SecurityError(f"Unauthorized import attempt: {alias.name}")
self.generic_visit(node)
def visit_ImportFrom(self, node: ast.ImportFrom):
if node.module:
base_module = node.module.split(".")[0]
if base_module not in self.ALLOWED_MODULES:
raise SecurityError(f"Unauthorized from-import attempt: {node.module}")
self.generic_visit(node)
def visit_Call(self, node: ast.Call):
if isinstance(node.func, ast.Name) and node.func.id in self.DISALLOWED_FUNCTIONS:
raise SecurityError(f"Disallowed call in schema AST: {node.func.id}()")
self.generic_visit(node)
class SecurityError(Exception):
pass
def validate_schema_ast(code_str: str) -> None:
tree = ast.parse(code_str)
validator = SchemaSecurityVisitor()
validator.visit(tree)
# ---------------------------------------------------------------------------
# Markdown Audit Logger for Arbitrary Schemas
# ---------------------------------------------------------------------------
class DynamicAuditLogger:
def __init__(self, output_path: str = "dynamic_debate_audit_report.md"):
self.output_path = output_path
self.entries: List[str] = []
self._init_log()
def _init_log(self):
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
self.entries.append("# Dynamic Multi-Model Engineering Debate & Audit Trail\n")
self.entries.append(f"**Execution Timestamp:** `{timestamp}` \n")
self.entries.append("---\n")
def log_problem(self, problem: str, constraints: str):
self.entries.append("## 1. Problem Statement & Governing Contradictions\n")
self.entries.append("```text")
self.entries.append(f"PROBLEM:\n{problem.strip()}\n\nCONSTRAINTS & TRADE-OFFS:\n{constraints.strip()}")
self.entries.append("```\n")
def log_generated_schemas(self, code_str: str):
self.entries.append("## 2. Dynamically Compiled Domain Schemas\n")
self.entries.append("```python")
self.entries.append(code_str.strip())
self.entries.append("```\n")
def log_initial_proposal(self, proposal: BaseModel):
self.entries.append("## 3. Iteration 0: Initial Candidate Design\n")
self.entries.append("```json")
self.entries.append(proposal.model_dump_json(indent=2))
self.entries.append("```\n")
def log_audit_round(self, round_num: int, audit: BaseModel):
self.entries.append(f"### Audit Round {round_num}: Adversarial Physical/System Review\n")
audit_dict = audit.model_dump()
has_critical = audit_dict.get("has_critical_defects", False)
status = "**REJECTED (Critical Defects Found)**" if has_critical else "**ACCEPTED (Clean)**"
self.entries.append(f"**Verdict:** {status} \n")
self.entries.append(f"**Summary:** {audit_dict.get('critique_summary', 'N/A')}\n")
defects = audit_dict.get("identified_defects", [])
if defects:
self.entries.append("| Severity | Category | Description | Recommended Mitigation |")
self.entries.append("|:---|:---|:---|:---|")
for d in defects:
sev = d.get("severity", "UNKNOWN")
cat = d.get("category", "General")
desc = d.get("description", "")
mit = d.get("recommended_mitigation", "")
self.entries.append(f"| `{sev}` | {cat} | {desc} | {mit} |")
self.entries.append("\n")
def log_iteration_diff(self, round_num: int, prev: BaseModel, new: BaseModel):
self.entries.append(f"## Iteration {round_num}: Architectural Revision & Structural Diff\n")
prev_lines = prev.model_dump_json(indent=2).splitlines(keepends=True)
new_lines = new.model_dump_json(indent=2).splitlines(keepends=True)
diff = list(difflib.unified_diff(
prev_lines,
new_lines,
fromfile=f"Proposal_Rev{round_num-1}.json",
tofile=f"Proposal_Rev{round_num}.json",
lineterm=""
))
self.entries.append("```diff")
if diff:
self.entries.extend([line.rstrip() for line in diff])
else:
self.entries.append("No structural differences detected.")
self.entries.append("```\n")
def log_final_specification(self, final_spec: BaseModel):
self.entries.append("## 4. Final Reconciled Production Specification\n")
self.entries.append("```json")
self.entries.append(final_spec.model_dump_json(indent=2))
self.entries.append("```\n")
def write_to_disk(self):
with open(self.output_path, "w", encoding="utf-8") as f:
f.write("\n".join(self.entries))
print(f"Audit log saved: {os.path.abspath(self.output_path)}")
# ---------------------------------------------------------------------------
# Fully Dynamic Multi-Model Engine
# ---------------------------------------------------------------------------
class DynamicMultiModelDebateEngine:
def __init__(self, api_key: Optional[str] = None, log_file: str = "dynamic_debate_audit_report.md"):
self.client = genai.Client(api_key=api_key or os.environ.get("GEMINI_API_KEY"))
self.architect_model = "gemini-2.5-pro"
self.critic_model = "gemini-2.5-flash"
self.synthesizer_model = "gemini-2.5-pro"
self.logger = DynamicAuditLogger(output_path=log_file)
def _call_structured(self, model: str, system_instruction: str, prompt: str, schema: Type[BaseModel]) -> BaseModel:
response = self.client.models.generate_content(
model=model,
contents=prompt,
config=types.GenerateContentConfig(
system_instruction=system_instruction,
response_mime_type="application/json",
response_schema=schema,
temperature=0.2,
),
)
return schema.model_validate_json(response.text)
def synthesize_domain_schemas(self, problem: str, constraints: str) -> Dict[str, Type[BaseModel]]:
"""Synthesizes, verifies, and compiles domain-specific Pydantic v2 schemas at runtime."""
meta_prompt = f"""
You are an expert Systems Metaprogrammer and Computational Ontologist specializing in Pydantic v2.
Generate a self-contained Python module defining four Pydantic v2 models tailored strictly to the domain physics and engineering contradictions of this problem:
Problem Statement:
{problem}
Governing Constraints & Trade-Offs:
{constraints}
Required Schema Class Names:
1. `DomainProposalSchema`:
- Contains precise domain-specific fields (e.g., fluid dynamics, biomechanics, electrical density, structural mass ratios).
- Uses `pydantic.Field` with explicit bounds (`ge`, `le`, `gt`) and physical units in descriptions.
- Includes state invariants and explicit subsystem allocations. Avoid generic labels.
2. `DomainDefectSchema`:
- Categorizes physical failure modes, parasitic overheads, and boundary violation specifics for this domain.
- Fields: `category: str`, `severity: str` ('CRITICAL', 'HIGH', 'MEDIUM', 'LOW'), `description: str`, `recommended_mitigation: str`.
3. `DomainAuditReportSchema`:
- `has_critical_defects: bool`
- `identified_defects: List[DomainDefectSchema]`
- `critique_summary: str`
4. `DomainFinalSpecificationSchema`:
- Reconciled engineering parameters, proven physical invariants, and explicit `residual_risks: List[str]`.
Rules:
- Valid imports only from: `pydantic`, `typing`, `typing_extensions`.
- Include `from pydantic import BaseModel, Field` and `from typing import List, Optional, Dict`.
- Return ONLY the executable code inside a single ```python ``` code block.
"""
response = self.client.models.generate_content(
model=self.architect_model,
contents=meta_prompt,
config=types.GenerateContentConfig(temperature=0.1)
)
match = re.search(r"```python\s*(.*?)\s*```", response.text, re.DOTALL)
code_str = match.group(1) if match else response.text
# 1. AST Safety Verification
validate_schema_ast(code_str)
# 2. In-memory dynamic compilation in an isolated namespace
namespace: Dict[str, Any] = {}
compiled_code = compile(code_str, filename="<dynamic_schemas>", mode="exec")
exec(compiled_code, namespace)
# 3. Extract and verify required models
required_classes = [
"DomainProposalSchema",
"DomainDefectSchema",
"DomainAuditReportSchema",
"DomainFinalSpecificationSchema"
]
extracted_schemas = {}
for cls_name in required_classes:
target = namespace.get(cls_name)
if not target or not (isinstance(target, type) and issubclass(target, BaseModel)):
raise ValueError(f"Dynamic schema compilation failed: Missing required class '{cls_name}'")
extracted_schemas[cls_name] = target
self.logger.log_generated_schemas(code_str)
return extracted_schemas
def run_debate(self, problem: str, constraints: str, max_iterations: int = 2) -> BaseModel:
self.logger.log_problem(problem, constraints)
print("[1/5] Synthesizing domain-specific Pydantic schemas via Metaprogrammer...")
schemas = self.synthesize_domain_schemas(problem, constraints)
ProposalCls = schemas["DomainProposalSchema"]
AuditCls = schemas["DomainAuditReportSchema"]
FinalCls = schemas["DomainFinalSpecificationSchema"]
print(" Compiled models: " + ", ".join(schemas.keys()))
print("\n[2/5] Generating initial domain candidate proposal...")
gen_sys_inst = (
"You are a Principal Engineering Systems Architect. Formulate an uncompromised, "
"rigorous engineering proposal that strictly resolves the stated contradictions. "
"Adhere strictly to the requested schema. Quantify physical parameters."
)
gen_prompt = f"Problem:\n{problem}\n\nConstraints:\n{constraints}\n\nPropose a complete technical solution."
proposal = self._call_structured(self.architect_model, gen_sys_inst, gen_prompt, ProposalCls)
self.logger.log_initial_proposal(proposal)
audit_history: List[BaseModel] = []
for round_idx in range(1, max_iterations + 1):
print(f"\n[3/5] Executing adversarial audit round {round_idx} of {max_iterations}...")
critic_sys_inst = (
"You are an Adversarial Peer Reviewer, Red-Teamer, and Physics Auditor. "
"Scrutinize the proposal strictly against governing physical laws, second-order parasitic "
"penalties, unstated assumptions, and fatigue/thermal/boundary failure modes. "
"Flag ungrounded optimism or unphysical 'free lunch' claims. Set has_critical_defects=True "
"if any fundamental physical, safety, or functional constraint is violated."
)
audit_prompt = (
f"Problem Statement:\n{problem}\n\n"
f"Governing Constraints:\n{constraints}\n\n"
f"Candidate Design Proposal:\n{proposal.model_dump_json(indent=2)}\n\n"
f"Perform the adversarial audit."
)
audit = self._call_structured(self.critic_model, critic_sys_inst, audit_prompt, AuditCls)
audit_history.append(audit)
self.logger.log_audit_round(round_idx, audit)
audit_dict = audit.model_dump()
defects = audit_dict.get("identified_defects", [])
critical_count = sum(1 for d in defects if d.get("severity") in ("CRITICAL", "HIGH"))
has_crit = audit_dict.get("has_critical_defects", False)
print(f" Identified {len(defects)} defects ({critical_count} critical/high).")
if not has_crit or round_idx == max_iterations:
if not has_crit:
print(" Passed audit: All critical trade-offs successfully resolved.")
else:
print(" Max iterations reached. Proceeding to final synthesis with documented residual risks.")
break
print(f"\n[4/5] Rerouting defects to Architect for iteration {round_idx + 1}...")
rev_sys_inst = (
"You are the Principal Systems Architect addressing hostile audit findings. "
"Directly remediate all flagged physical failure modes and parasitic trade-offs "
"by adjusting dimensions, material specs, or operational parameters."
)
defects_summary = "\n".join([
f"- [{d.get('severity')}] {d.get('category')}: {d.get('description')} -> Fix: {d.get('recommended_mitigation')}"
for d in defects
])
rev_prompt = (
f"Problem & Constraints:\n{problem}\n{constraints}\n\n"
f"Prior Proposal:\n{proposal.model_dump_json(indent=2)}\n\n"
f"Auditor Defects to Resolve:\n{defects_summary}\n\n"
f"Produce a revised architectural proposal."
)
revised_proposal = self._call_structured(self.architect_model, rev_sys_inst, rev_prompt, ProposalCls)
self.logger.log_iteration_diff(round_idx, proposal, revised_proposal)
proposal = revised_proposal
print("\n[5/5] Synthesizing final technical specification...")
synth_sys_inst = (
"You are the Executive Technical Authority. Reconcile the design history, final iteration, "
"and audit findings into an authoritative production specification. Convert unresolved edge-cases "
"into documented residual operational risks."
)
history_summaries = [f"Round {i}: {a.model_dump().get('critique_summary')}" for i, a in enumerate(audit_history, 1)]
synth_prompt = (
f"Problem & Constraints:\n{problem}\n{constraints}\n\n"
f"Vetted Proposal:\n{proposal.model_dump_json(indent=2)}\n\n"
f"Audit Iteration History:\n" + "\n".join(history_summaries) + "\n\n"
f"Produce the final reconciled technical specification."
)
final_spec = self._call_structured(self.synthesizer_model, synth_sys_inst, synth_prompt, FinalCls)
self.logger.log_final_specification(final_spec)
self.logger.write_to_disk()
print("Engine run complete.\n")
return final_spec
# ---------------------------------------------------------------------------
# Execution Example
# ---------------------------------------------------------------------------
if __name__ == "__main__":
# Test on an arbitrary coupled physical engineering contradiction
problem = (
"Increase the gravimetric and volumetric energy density of an electric vehicle battery pack "
"system while maintaining baseline system weight, retaining zero cell-to-cell thermal propagation, "
"and not increasing pack unit production cost ($/kWh)."
)
constraints = (
"- Silicon anode volumetric swelling (>100% lithiation expansion) induces mechanical stress.\n"
"- High-nickel cathodes (e.g. NMC 811) increase runaway exothermic reaction kinetics.\n"
"- Total pack mass budget <= 450 kg; Pack volume <= 320 L.\n"
"- Unit cost must remain <= $95/kWh at the finished pack level.\n"
"- Passive propagation resistance (UL 9540A compliant): single-cell runaway must not breach adjacent cells."
)
engine = DynamicMultiModelDebateEngine(log_file="ev_battery_dynamic_debate.md")
result = engine.run_debate(problem, constraints, max_iterations=2)
print("=== FINAL SPECIFICATION OUTPUT ===")
print(result.model_dump_json(indent=2))
Pipeline Mechanics
Autonomous Metaprogramming: The model inspects the raw engineering problem and constraints, deciding which parameters (e.g., cell swelling limits, thermal conductivity, $/kWh) are essential to track as strongly-typed attributes.
AST Gating: Before executing the generated code via
exec(),validate_schema_ast()parses the AST to verify that the generated code contains zero external system calls, file operations, unauthorized network libraries, or shell injections.Dynamic Compilation: Models are compiled into an isolated dictionary namespace and referenced as standard Python types (
Type[BaseModel]).Structured Native API Enforcement: The dynamically generated types are passed directly into Gemini's
types.GenerateContentConfig(response_schema=schema), enforcing JSON schema compliance at the decoding level across both the Generator and the Auditor.
Update DynamicMultiModelDebateEngine with an automatic repair loop that feeds AST validation or compilation errors back to the metaprogrammer for self-correction.
Here is the updated implementation. It wraps schema synthesis in an automated Generator–Grader–Improver self-correction loop (synthesize_domain_schemas_with_repair).
If the generated code trips an ast.PyCF_ONLY_AST syntax error, violates the AST security policy, fails Python bytecode compilation, or omits any of the four mandatory Pydantic class definitions, the caught exception and stack trace are fed back into the metaprogrammer prompt to trigger targeted self-repair across up to max_schema_attempts.
import os
import re
import ast
import json
import difflib
from datetime import datetime
from typing import List, Dict, Any, Type, Optional
from pydantic import BaseModel
from google import genai
from google.genai import types
# ---------------------------------------------------------------------------
# AST Security Guardrails
# ---------------------------------------------------------------------------
class SecurityError(Exception):
pass
class SchemaCompilationError(Exception):
pass
class SchemaSecurityVisitor(ast.NodeVisitor):
"""
Enforces that dynamically generated code contains strictly declarative Pydantic schemas,
blocking file I/O, subprocesses, network calls, eval/exec, and unauthorized system imports.
"""
ALLOWED_MODULES = {"pydantic", "typing", "typing_extensions", "math", "datetime", "enum"}
DISALLOWED_FUNCTIONS = {"exec", "eval", "compile", "open", "__import__", "getattr", "setattr"}
def visit_Import(self, node: ast.Import):
for alias in node.names:
base_module = alias.name.split(".")[0]
if base_module not in self.ALLOWED_MODULES:
raise SecurityError(f"Unauthorized import attempt: '{alias.name}'. Only {self.ALLOWED_MODULES} are permitted.")
self.generic_visit(node)
def visit_ImportFrom(self, node: ast.ImportFrom):
if node.module:
base_module = node.module.split(".")[0]
if base_module not in self.ALLOWED_MODULES:
raise SecurityError(f"Unauthorized from-import attempt: '{node.module}'. Only {self.ALLOWED_MODULES} are permitted.")
self.generic_visit(node)
def visit_Call(self, node: ast.Call):
if isinstance(node.func, ast.Name) and node.func.id in self.DISALLOWED_FUNCTIONS:
raise SecurityError(f"Disallowed function call detected in schema AST: '{node.func.id}()'")
self.generic_visit(node)
def validate_and_compile_schema(code_str: str) -> Dict[str, Type[BaseModel]]:
"""
Validates AST syntax, security constraints, and bytecode compilation in an isolated namespace.
Raises SchemaCompilationError or SecurityError on failure.
"""
# 1. Parse AST (checks for valid Python syntax)
try:
tree = ast.parse(code_str, filename="<dynamic_schemas>")
except SyntaxError as e:
raise SchemaCompilationError(f"SyntaxError on line {e.lineno}: {e.msg}") from e
# 2. Run AST Security Audit
visitor = SchemaSecurityVisitor()
visitor.visit(tree)
# 3. Compile bytecode
try:
compiled_code = compile(tree, filename="<dynamic_schemas>", mode="exec")
except Exception as e:
raise SchemaCompilationError(f"Bytecode compilation failed: {str(e)}") from e
# 4. Execute in an isolated namespace
namespace: Dict[str, Any] = {}
try:
exec(compiled_code, namespace)
except Exception as e:
raise SchemaCompilationError(f"Namespace execution failed: {type(e).__name__}: {str(e)}") from e
# 5. Verify presence and validity of required models
required_classes = [
"DomainProposalSchema",
"DomainDefectSchema",
"DomainAuditReportSchema",
"DomainFinalSpecificationSchema"
]
extracted_schemas = {}
for cls_name in required_classes:
target = namespace.get(cls_name)
if not target or not (isinstance(target, type) and issubclass(target, BaseModel)):
raise SchemaCompilationError(
f"Missing or invalid class '{cls_name}'. Must be a defined subclass of pydantic.BaseModel."
)
extracted_schemas[cls_name] = target
return extracted_schemas
# ---------------------------------------------------------------------------
# Dynamic Markdown Audit Logger
# ---------------------------------------------------------------------------
class DynamicAuditLogger:
def __init__(self, output_path: str = "dynamic_debate_audit_report.md"):
self.output_path = output_path
self.entries: List[str] = []
self._init_log()
def _init_log(self):
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
self.entries.append("# Dynamic Multi-Model Engineering Debate & Audit Trail\n")
self.entries.append(f"**Execution Timestamp:** `{timestamp}` \n")
self.entries.append("---\n")
def log_problem(self, problem: str, constraints: str):
self.entries.append("## 1. Problem Statement & Governing Contradictions\n")
self.entries.append("```text")
self.entries.append(f"PROBLEM:\n{problem.strip()}\n\nCONSTRAINTS & TRADE-OFFS:\n{constraints.strip()}")
self.entries.append("```\n")
def log_schema_repair_event(self, attempt: int, error_type: str, error_message: str):
self.entries.append(f"### ⚠️ Schema Compilation Alert (Attempt {attempt})\n")
self.entries.append(f"**Error Category:** `{error_type}` \n")
self.entries.append(f"**Diagnostic Trace:**\n```text\n{error_message.strip()}\n```\n")
def log_generated_schemas(self, code_str: str, attempts: int):
self.entries.append(f"## 2. Verified Domain Schemas (Synthesized in {attempts} attempt(s))\n")
self.entries.append("```python")
self.entries.append(code_str.strip())
self.entries.append("```\n")
def log_initial_proposal(self, proposal: BaseModel):
self.entries.append("## 3. Iteration 0: Initial Candidate Design\n")
self.entries.append("```json")
self.entries.append(proposal.model_dump_json(indent=2))
self.entries.append("```\n")
def log_audit_round(self, round_num: int, audit: BaseModel):
self.entries.append(f"### Audit Round {round_num}: Adversarial Physical/System Review\n")
audit_dict = audit.model_dump()
has_critical = audit_dict.get("has_critical_defects", False)
status = "**REJECTED (Critical Defects Found)**" if has_critical else "**ACCEPTED (Clean)**"
self.entries.append(f"**Verdict:** {status} \n")
self.entries.append(f"**Summary:** {audit_dict.get('critique_summary', 'N/A')}\n")
defects = audit_dict.get("identified_defects", [])
if defects:
self.entries.append("| Severity | Category | Description | Recommended Mitigation |")
self.entries.append("|:---|:---|:---|:---|")
for d in defects:
sev = d.get("severity", "UNKNOWN")
cat = d.get("category", "General")
desc = d.get("description", "")
mit = d.get("recommended_mitigation", "")
self.entries.append(f"| `{sev}` | {cat} | {desc} | {mit} |")
self.entries.append("\n")
def log_iteration_diff(self, round_num: int, prev: BaseModel, new: BaseModel):
self.entries.append(f"## Iteration {round_num}: Architectural Revision & Structural Diff\n")
prev_lines = prev.model_dump_json(indent=2).splitlines(keepends=True)
new_lines = new.model_dump_json(indent=2).splitlines(keepends=True)
diff = list(difflib.unified_diff(
prev_lines,
new_lines,
fromfile=f"Proposal_Rev{round_num-1}.json",
tofile=f"Proposal_Rev{round_num}.json",
lineterm=""
))
self.entries.append("```diff")
if diff:
self.entries.extend([line.rstrip() for line in diff])
else:
self.entries.append("No structural differences detected.")
self.entries.append("```\n")
def log_final_specification(self, final_spec: BaseModel):
self.entries.append("## 4. Final Reconciled Production Specification\n")
self.entries.append("```json")
self.entries.append(final_spec.model_dump_json(indent=2))
self.entries.append("```\n")
def write_to_disk(self):
with open(self.output_path, "w", encoding="utf-8") as f:
f.write("\n".join(self.entries))
print(f"Audit log saved: {os.path.abspath(self.output_path)}")
# ---------------------------------------------------------------------------
# Dynamic Engine with Automatic Schema Repair Loop
# ---------------------------------------------------------------------------
class DynamicMultiModelDebateEngine:
def __init__(self, api_key: Optional[str] = None, log_file: str = "dynamic_debate_audit_report.md"):
self.client = genai.Client(api_key=api_key or os.environ.get("GEMINI_API_KEY"))
self.architect_model = "gemini-2.5-pro"
self.critic_model = "gemini-2.5-flash"
self.synthesizer_model = "gemini-2.5-pro"
self.logger = DynamicAuditLogger(output_path=log_file)
def _call_structured(self, model: str, system_instruction: str, prompt: str, schema: Type[BaseModel]) -> BaseModel:
response = self.client.models.generate_content(
model=model,
contents=prompt,
config=types.GenerateContentConfig(
system_instruction=system_instruction,
response_mime_type="application/json",
response_schema=schema,
temperature=0.2,
),
)
return schema.model_validate_json(response.text)
def _extract_python_code(self, raw_text: str) -> str:
match = re.search(r"```python\s*(.*?)\s*```", raw_text, re.DOTALL)
return match.group(1).strip() if match else raw_text.strip()
def synthesize_domain_schemas_with_repair(
self, problem: str, constraints: str, max_schema_attempts: int = 3
) -> Dict[str, Type[BaseModel]]:
"""
Synthesizes domain Pydantic schemas with an automated self-correction loop.
Catches AST security breaches, syntax errors, and missing classes, feeding diagnostics back to the LLM.
"""
meta_prompt = f"""You are an expert Systems Metaprogrammer and Computational Ontologist specializing in Pydantic v2.
Generate a self-contained Python module defining four Pydantic v2 models tailored strictly to the domain physics and engineering contradictions of this problem:
Problem Statement:
{problem}
Governing Constraints & Trade-Offs:
{constraints}
Required Schema Class Names:
1. `DomainProposalSchema`:
- Contains precise domain-specific fields (e.g., fluid dynamics, biomechanics, electrical density, structural mass ratios).
- Uses `pydantic.Field` with explicit bounds (`ge`, `le`, `gt`) and physical units in descriptions.
- Includes state invariants and explicit subsystem allocations. Avoid generic labels.
2. `DomainDefectSchema`:
- Categorizes physical failure modes, parasitic overheads, and boundary violation specifics for this domain.
- Fields: `category: str`, `severity: str` ('CRITICAL', 'HIGH', 'MEDIUM', 'LOW'), `description: str`, `recommended_mitigation: str`.
3. `DomainAuditReportSchema`:
- `has_critical_defects: bool`
- `identified_defects: List[DomainDefectSchema]`
- `critique_summary: str`
4. `DomainFinalSpecificationSchema`:
- Reconciled engineering parameters, proven physical invariants, and explicit `residual_risks: List[str]`.
Strict Architectural & Security Rules:
- Valid imports only from: `pydantic`, `typing`, `typing_extensions`, `math`.
- Absolutely NO unauthorized imports (`os`, `sys`, `subprocess`, `requests`, `socket`, etc.).
- Absolutely NO executable function calls at the module level (`open()`, `eval()`, `exec()`).
- Return ONLY the executable Python code enclosed within a single ```python ... ``` block.
"""
current_prompt = meta_prompt
last_generated_code = ""
for attempt in range(1, max_schema_attempts + 1):
print(f" [Metaprogrammer Attempt {attempt}/{max_schema_attempts}] Synthesizing schemas...")
response = self.client.models.generate_content(
model=self.architect_model,
contents=current_prompt,
config=types.GenerateContentConfig(temperature=0.1)
)
code_str = self._extract_python_code(response.text)
last_generated_code = code_str
try:
# Run validation, AST security inspection, and bytecode execution
compiled_schemas = validate_and_compile_schema(code_str)
print(f" [Success] Schemas passed AST security and compile checks on attempt {attempt}.")
self.logger.log_generated_schemas(code_str, attempts=attempt)
return compiled_schemas
except (SecurityError, SchemaCompilationError) as err:
error_category = type(err).__name__
error_message = str(err)
print(f" [Warning] Attempt {attempt} failed ({error_category}): {error_message}")
self.logger.log_schema_repair_event(attempt, error_category, error_message)
if attempt == max_schema_attempts:
raise RuntimeError(
f"Metaprogrammer failed to synthesize valid schemas after {max_schema_attempts} attempts. "
f"Last error: {error_message}"
) from err
# Formulate targeted self-repair prompt
current_prompt = f"""Your previous Pydantic v2 code submission failed validation and could not be loaded.
Diagnostic Error ({error_category}):
{error_message}
Rejected Code:
```python
{code_str}
Correction Directives:
Fix the specific error identified above.
If an import failed or violated security policy, remove it and use standard
typingorpydanticprimitives.Ensure all 4 required classes (
DomainProposalSchema,DomainDefectSchema,DomainAuditReportSchema,DomainFinalSpecificationSchema) are present and inherit frompydantic.BaseModel.Output the corrected code inside a single
python ...block. """raise RuntimeError("Unexpected exit from schema synthesis loop.")def run_debate(self, problem: str, constraints: str, max_iterations: int = 2) -> BaseModel: self.logger.log_problem(problem, constraints)
print("[1/5] Synthesizing and verifying domain schemas via Metaprogrammer self-repair loop...") schemas = self.synthesize_domain_schemas_with_repair(problem, constraints, max_schema_attempts=3) ProposalCls = schemas["DomainProposalSchema"] AuditCls = schemas["DomainAuditReportSchema"] FinalCls = schemas["DomainFinalSpecificationSchema"] print("\n[2/5] Generating initial candidate proposal...") gen_sys_inst = ( "You are a Principal Engineering Systems Architect. Formulate an uncompromised, " "rigorous engineering proposal that strictly resolves the stated contradictions. " "Adhere strictly to the requested schema. Quantify physical parameters." ) gen_prompt = f"Problem:\n{problem}\n\nConstraints:\n{constraints}\n\nPropose a complete technical solution." proposal = self._call_structured(self.architect_model, gen_sys_inst, gen_prompt, ProposalCls) self.logger.log_initial_proposal(proposal) audit_history: List[BaseModel] = [] for round_idx in range(1, max_iterations + 1): print(f"\n[3/5] Executing adversarial audit round {round_idx} of {max_iterations}...") critic_sys_inst = ( "You are an Adversarial Peer Reviewer, Red-Teamer, and Physics Auditor. " "Scrutinize the proposal strictly against governing physical laws, second-order parasitic " "penalties, unstated assumptions, and fatigue/thermal/boundary failure modes. " "Flag ungrounded optimism or unphysical 'free lunch' claims. Set has_critical_defects=True " "if any fundamental physical, safety, or functional constraint is violated." ) audit_prompt = ( f"Problem Statement:\n{problem}\n\n" f"Governing Constraints:\n{constraints}\n\n" f"Candidate Design Proposal:\n{proposal.model_dump_json(indent=2)}\n\n" f"Perform the adversarial audit." ) audit = self._call_structured(self.critic_model, critic_sys_inst, audit_prompt, AuditCls) audit_history.append(audit) self.logger.log_audit_round(round_idx, audit) audit_dict = audit.model_dump() defects = audit_dict.get("identified_defects", []) critical_count = sum(1 for d in defects if d.get("severity") in ("CRITICAL", "HIGH")) has_crit = audit_dict.get("has_critical_defects", False) print(f" Identified {len(defects)} defects ({critical_count} critical/high).") if not has_crit or round_idx == max_iterations: if not has_crit: print(" Passed audit: All critical trade-offs successfully resolved.") else: print(" Max iterations reached. Proceeding to final synthesis with documented residual risks.") break print(f"\n[4/5] Rerouting defects to Architect for iteration {round_idx + 1}...") rev_sys_inst = ( "You are the Principal Systems Architect addressing hostile audit findings. " "Directly remediate all flagged physical failure modes and parasitic trade-offs " "by adjusting dimensions, material specs, or operational parameters." ) defects_summary = "\n".join([ f"- [{d.get('severity')}] {d.get('category')}: {d.get('description')} -> Fix: {d.get('recommended_mitigation')}" for d in defects ]) rev_prompt = ( f"Problem & Constraints:\n{problem}\n{constraints}\n\n" f"Prior Proposal:\n{proposal.model_dump_json(indent=2)}\n\n" f"Auditor Defects to Resolve:\n{defects_summary}\n\n" f"Produce a revised architectural proposal." ) revised_proposal = self._call_structured(self.architect_model, rev_sys_inst, rev_prompt, ProposalCls) self.logger.log_iteration_diff(round_idx, proposal, revised_proposal) proposal = revised_proposal print("\n[5/5] Synthesizing final technical specification...") synth_sys_inst = ( "You are the Executive Technical Authority. Reconcile the design history, final iteration, " "and audit findings into an authoritative production specification. Convert unresolved edge-cases " "into documented residual operational risks." ) history_summaries = [f"Round {i}: {a.model_dump().get('critique_summary')}" for i, a in enumerate(audit_history, 1)] synth_prompt = ( f"Problem & Constraints:\n{problem}\n{constraints}\n\n" f"Vetted Proposal:\n{proposal.model_dump_json(indent=2)}\n\n" f"Audit Iteration History:\n" + "\n".join(history_summaries) + "\n\n" f"Produce the final reconciled technical specification." ) final_spec = self._call_structured(self.synthesizer_model, synth_sys_inst, synth_prompt, FinalCls) self.logger.log_final_specification(final_spec) self.logger.write_to_disk() print("Engine run complete.\n") return final_spec
---------------------------------------------------------------------------
Execution Entry Point
---------------------------------------------------------------------------
if name == "main":
problem = (
"Improve the aerodynamic efficiency of an endurance racing bicycle while preserving " "rider pedaling power output and biomechanical comfort over a 6-hour duration."
)
constraints = (
"- Air drag power scales with CdA and v^3 (v = 45 km/h benchmark).\n"
"- Closing torso/hip angle below 45 degrees at Top Dead Center restricts breathing tidal volume and glute recruitment.\n"
"- Extreme forward head reach creates cervical spine fatigue exceeding endurance threshold.\n"
"- Lateral fork handling stability must tolerate 25 km/h crosswinds at 15-degree yaw angles."
)
engine = DynamicMultiModelDebateEngine(log_file="bike_aerodynamics_debate.md")
final_design = engine.run_debate(problem, constraints, max_iterations=2)
---
### Key Self-Correction Improvements
* **Triaged Failure Handling:** Separates syntax failures (`SyntaxError`), unauthorized system calls (`SecurityError`), bytecode compilation issues, and schema structure errors (`SchemaCompilationError`).
* **Differential Diagnostic Feedback:** Instead of re-issuing the raw prompt, the repair step hands the model the failed code alongside the exact Python exception line number and reason, directing it to repair the specific flaw without discarding the domain-specific parameters it already reasoned through.
* **Audit Trail Integration:** Failed schema attempts and repair events are recorded directly into the output Markdown report under a dedicated diagnostic section before the verified schema definition.
Want to unit test the repair loop by deliberately feeding malformed and insecure schema prompts to verify the repair mechanism?" query="Write a pytest test suite that deliberately feeds invalid syntax, disallowed imports, and missing classes to verify the schema repair loop."/>
To extend the metaprogramming and debate pipeline:
Add SymPy equation validation to the self-repair loop
query="Integrate SymPy equation checking into the schema validation loop so physical laws (like aerodynamic power equations) are formally verified before the debate begins."
Persist verified dynamic schemas to an on-disk schema cache
query="Add a disk-based schema cache to the engine that hashes problem statements and skips metaprogramming when an identical or similar problem has already been compiled."
Integrate SymPy equation checking into the schema validation loop so physical laws (like aerodynamic power equations) are formally verified before the debate begins.
Here is the integrated pipeline. It inserts a deterministic SymPy Mathematical Verification Stage directly into the dynamic schema compilation and self-repair loop.
Architectural Mechanism
Mathematical Invariant Directives: The metaprogrammer is instructed to synthesize a dedicated
PhysicsVerificationSandboxclass containing symbolic representations of the governing equations (e.g., aerodynamic drag power P = 1/2 rho v^3 C_d A, battery volumetric packaging efficiency eta_v, or torque/cadence relations), along with symbolic boundary checks (e.g., P > 0 for v > 0, sign of {partial P} / {partial C_d A} > 0).AST Gating Expansion: The AST validator's permitted imports are expanded to include
sympywhile strictly blocking execution primitives (eval,exec, arbitrary file/system calls).Symbolic Sanity Sandbox: Before the debate can proceed,
validate_and_compile_schemainvokes the compiledPhysicsVerificationSandbox.verify_physical_invariants(). If an equation is mathematically contradictory, violates dimensional scaling, or contains undefined symbols, SymPy raises an exception that routes directly into the metaprogrammer's self-repair loop alongside any AST/syntax failures.
import os
import re
import ast
import json
import difflib
from datetime import datetime
from typing import List, Dict, Any, Type, Optional
from pydantic import BaseModel, Field
import sympy as sp
from google import genai
from google.genai import types
# ---------------------------------------------------------------------------
# AST Security Guardrails (Expanded with SymPy Whitelist)
# ---------------------------------------------------------------------------
class SecurityError(Exception):
pass
class SchemaCompilationError(Exception):
pass
class PhysicsVerificationError(Exception):
pass
class SchemaSecurityVisitor(ast.NodeVisitor):
"""
Enforces that dynamically generated code contains strictly declarative Pydantic schemas
and formal SymPy equation declarations, blocking file I/O, subprocesses, network calls,
and arbitrary system access.
"""
ALLOWED_MODULES = {"pydantic", "typing", "typing_extensions", "math", "datetime", "enum", "sympy"}
DISALLOWED_FUNCTIONS = {"exec", "eval", "compile", "open", "__import__", "getattr", "setattr"}
def visit_Import(self, node: ast.Import):
for alias in node.names:
base_module = alias.name.split(".")[0]
if base_module not in self.ALLOWED_MODULES:
raise SecurityError(f"Unauthorized import attempt: '{alias.name}'. Only {self.ALLOWED_MODULES} are permitted.")
self.generic_visit(node)
def visit_ImportFrom(self, node: ast.ImportFrom):
if node.module:
base_module = node.module.split(".")[0]
if base_module not in self.ALLOWED_MODULES:
raise SecurityError(f"Unauthorized from-import attempt: '{node.module}'. Only {self.ALLOWED_MODULES} are permitted.")
self.generic_visit(node)
def visit_Call(self, node: ast.Call):
if isinstance(node.func, ast.Name) and node.func.id in self.DISALLOWED_FUNCTIONS:
raise SecurityError(f"Disallowed function call detected in schema AST: '{node.func.id}()'")
self.generic_visit(node)
def validate_and_compile_schema(code_str: str) -> Dict[str, Any]:
"""
Validates AST syntax, security constraints, bytecode compilation in an isolated namespace,
and executes SymPy symbolic physics verification before admitting schemas to the debate.
"""
# 1. Parse AST
try:
tree = ast.parse(code_str, filename="<dynamic_schemas>")
except SyntaxError as e:
raise SchemaCompilationError(f"SyntaxError on line {e.lineno}: {e.msg}") from e
# 2. Run AST Security Audit
visitor = SchemaSecurityVisitor()
visitor.visit(tree)
# 3. Compile bytecode
try:
compiled_code = compile(tree, filename="<dynamic_schemas>", mode="exec")
except Exception as e:
raise SchemaCompilationError(f"Bytecode compilation failed: {str(e)}") from e
# 4. Execute in an isolated namespace
namespace: Dict[str, Any] = {}
try:
exec(compiled_code, namespace)
except Exception as e:
raise SchemaCompilationError(f"Namespace execution failed: {type(e).__name__}: {str(e)}") from e
# 5. Verify presence of required models and Sandbox
required_classes = [
"DomainProposalSchema",
"DomainDefectSchema",
"DomainAuditReportSchema",
"DomainFinalSpecificationSchema",
"PhysicsVerificationSandbox"
]
extracted: Dict[str, Any] = {}
for cls_name in required_classes:
target = namespace.get(cls_name)
if not target:
raise SchemaCompilationError(f"Missing required class '{cls_name}'.")
if cls_name != "PhysicsVerificationSandbox":
if not (isinstance(target, type) and issubclass(target, BaseModel)):
raise SchemaCompilationError(f"'{cls_name}' must inherit from pydantic.BaseModel.")
extracted[cls_name] = target
# 6. Execute SymPy Formal Physics Verification
sandbox_cls = extracted["PhysicsVerificationSandbox"]
if not hasattr(sandbox_cls, "verify_physical_invariants") or not callable(sandbox_cls.verify_physical_invariants):
raise SchemaCompilationError("PhysicsVerificationSandbox must define a callable staticmethod 'verify_physical_invariants()'.")
try:
verification_passed, report = sandbox_cls.verify_physical_invariants()
if not verification_passed:
raise PhysicsVerificationError(f"SymPy physics validation failed: {report}")
except Exception as e:
if isinstance(e, PhysicsVerificationError):
raise
raise PhysicsVerificationError(f"SymPy calculation crashed during formal verification: {type(e).__name__}: {str(e)}") from e
extracted["_physics_report"] = report
return extracted
# ---------------------------------------------------------------------------
# Dynamic Markdown Audit Logger with Physics Invariant Section
# ---------------------------------------------------------------------------
class DynamicAuditLogger:
def __init__(self, output_path: str = "dynamic_debate_audit_report.md"):
self.output_path = output_path
self.entries: List[str] = []
self._init_log()
def _init_log(self):
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
self.entries.append("# Dynamic Multi-Model Engineering Debate & Audit Trail\n")
self.entries.append(f"**Execution Timestamp:** `{timestamp}` \n")
self.entries.append("---\n")
def log_problem(self, problem: str, constraints: str):
self.entries.append("## 1. Problem Statement & Governing Contradictions\n")
self.entries.append("```text")
self.entries.append(f"PROBLEM:\n{problem.strip()}\n\nCONSTRAINTS & TRADE-OFFS:\n{constraints.strip()}")
self.entries.append("```\n")
def log_schema_repair_event(self, attempt: int, error_type: str, error_message: str):
self.entries.append(f"### ⚠️ Schema / Physics Verification Failure (Attempt {attempt})\n")
self.entries.append(f"**Error Category:** `{error_type}` \n")
self.entries.append(f"**Diagnostic Trace:**\n```text\n{error_message.strip()}\n```\n")
def log_generated_schemas(self, code_str: str, attempts: int, physics_report: str):
self.entries.append(f"## 2. Verified Domain Schemas & Formal SymPy Invariants\n")
self.entries.append(f"**Compilation Status:** Verified in {attempts} attempt(s) with formal mathematical proof. \n")
self.entries.append(f"**SymPy Verification Report:**\n> {physics_report}\n")
self.entries.append("```python")
self.entries.append(code_str.strip())
self.entries.append("```\n")
def log_initial_proposal(self, proposal: BaseModel):
self.entries.append("## 3. Iteration 0: Initial Candidate Design\n")
self.entries.append("```json")
self.entries.append(proposal.model_dump_json(indent=2))
self.entries.append("```\n")
def log_audit_round(self, round_num: int, audit: BaseModel):
self.entries.append(f"### Audit Round {round_num}: Adversarial Physical/System Review\n")
audit_dict = audit.model_dump()
has_critical = audit_dict.get("has_critical_defects", False)
status = "**REJECTED (Critical Defects Found)**" if has_critical else "**ACCEPTED (Clean)**"
self.entries.append(f"**Verdict:** {status} \n")
self.entries.append(f"**Summary:** {audit_dict.get('critique_summary', 'N/A')}\n")
defects = audit_dict.get("identified_defects", [])
if defects:
self.entries.append("| Severity | Category | Description | Recommended Mitigation |")
self.entries.append("|:---|:---|:---|:---|")
for d in defects:
sev = d.get("severity", "UNKNOWN")
cat = d.get("category", "General")
desc = d.get("description", "")
mit = d.get("recommended_mitigation", "")
self.entries.append(f"| `{sev}` | {cat} | {desc} | {mit} |")
self.entries.append("\n")
def log_iteration_diff(self, round_num: int, prev: BaseModel, new: BaseModel):
self.entries.append(f"## Iteration {round_num}: Architectural Revision & Structural Diff\n")
prev_lines = prev.model_dump_json(indent=2).splitlines(keepends=True)
new_lines = new.model_dump_json(indent=2).splitlines(keepends=True)
diff = list(difflib.unified_diff(
prev_lines,
new_lines,
fromfile=f"Proposal_Rev{round_num-1}.json",
tofile=f"Proposal_Rev{round_num}.json",
lineterm=""
))
self.entries.append("```diff")
if diff:
self.entries.extend([line.rstrip() for line in diff])
else:
self.entries.append("No structural differences detected.")
self.entries.append("```\n")
def log_final_specification(self, final_spec: BaseModel):
self.entries.append("## 4. Final Reconciled Production Specification\n")
self.entries.append("```json")
self.entries.append(final_spec.model_dump_json(indent=2))
self.entries.append("```\n")
def write_to_disk(self):
with open(self.output_path, "w", encoding="utf-8") as f:
f.write("\n".join(self.entries))
print(f"Audit log saved: {os.path.abspath(self.output_path)}")
# ---------------------------------------------------------------------------
# Dynamic Engine with SymPy Invariant Checker
# ---------------------------------------------------------------------------
class DynamicMultiModelDebateEngine:
def __init__(self, api_key: Optional[str] = None, log_file: str = "dynamic_debate_audit_report.md"):
self.client = genai.Client(api_key=api_key or os.environ.get("GEMINI_API_KEY"))
self.architect_model = "gemini-2.5-pro"
self.critic_model = "gemini-2.5-flash"
self.synthesizer_model = "gemini-2.5-pro"
self.logger = DynamicAuditLogger(output_path=log_file)
def _call_structured(self, model: str, system_instruction: str, prompt: str, schema: Type[BaseModel]) -> BaseModel:
response = self.client.models.generate_content(
model=model,
contents=prompt,
config=types.GenerateContentConfig(
system_instruction=system_instruction,
response_mime_type="application/json",
response_schema=schema,
temperature=0.2,
),
)
return schema.model_validate_json(response.text)
def _extract_python_code(self, raw_text: str) -> str:
match = re.search(r"```python\s*(.*?)\s*```", raw_text, re.DOTALL)
return match.group(1).strip() if match else raw_text.strip()
def synthesize_domain_schemas_with_physics(
self, problem: str, constraints: str, max_schema_attempts: int = 3
) -> Dict[str, Any]:
"""
Synthesizes domain Pydantic schemas and SymPy physical invariants with an automated repair loop.
Catches AST security breaches, syntax errors, and SymPy physical equation validation failures.
"""
meta_prompt = f"""You are an expert Systems Metaprogrammer, Computational Physicist, and Ontologist.
Generate a self-contained Python module defining four Pydantic v2 models and one SymPy Physics Verification Sandbox tailored strictly to the governing physical laws of this problem:
Problem Statement:
{problem}
Governing Constraints & Trade-Offs:
{constraints}
Required Schema Class Names & Specifications:
1. `DomainProposalSchema`:
- Contains precise domain-specific parameters (quantitative metrics with units).
- Uses `pydantic.Field` with explicit bounds (`ge`, `le`, `gt`).
- Quantifies the design state, invariants, and subsystem allocations.
2. `DomainDefectSchema`:
- Categorizes physical failure modes, parasitic overheads, and boundary violation specifics for this domain.
- Fields: `category: str`, `severity: str` ('CRITICAL', 'HIGH', 'MEDIUM', 'LOW'), `description: str`, `recommended_mitigation: str`.
3. `DomainAuditReportSchema`:
- `has_critical_defects: bool`
- `identified_defects: List[DomainDefectSchema]`
- `critique_summary: str`
4. `DomainFinalSpecificationSchema`:
- Reconciled engineering parameters, validated invariants, and explicit `residual_risks: List[str]`.
5. `PhysicsVerificationSandbox`:
- A standalone class implementing `@staticmethod def verify_physical_invariants() -> tuple[bool, str]:`
- Inside this method, define SymPy symbols (`sp.Symbol`) for all primary variables in the governing equations (e.g. power, drag, density, velocity, efficiency, mass).
- Formulate the primary governing equation(s) as SymPy expressions or relational equalities (e.g., aerodynamic power $P = \\frac{1}{2}\\rho v^3 C_d A$).
- Formally verify scaling derivatives or boundary conditions using SymPy calculus (e.g., `sp.diff(P, v)` must be positive for positive constants; verify conservation constraints or dimensional sanity).
- Return `(True, "Summary of verified equations and invariants")` on success, or `(False, "Explanation of physical contradiction")` if invariants fail.
Strict Architectural & Security Rules:
- Allowed imports ONLY: `pydantic`, `typing`, `typing_extensions`, `math`, `sympy as sp` (or `import sympy as sp`).
- Absolutely NO unauthorized imports (`os`, `sys`, `subprocess`, `requests`, `socket`, etc.).
- Absolutely NO executable module-level functions (`eval()`, `exec()`, `open()`).
- Return ONLY the executable Python code inside a single ```python ... ``` block.
"""
current_prompt = meta_prompt
for attempt in range(1, max_schema_attempts + 1):
print(f" [Metaprogrammer + SymPy Attempt {attempt}/{max_schema_attempts}] Synthesizing schemas...")
response = self.client.models.generate_content(
model=self.architect_model,
contents=current_prompt,
config=types.GenerateContentConfig(temperature=0.1)
)
code_str = self._extract_python_code(response.text)
try:
# Compile, verify AST, and run SymPy equation sandbox
compiled_bundle = validate_and_compile_schema(code_str)
physics_report = compiled_bundle["_physics_report"]
print(f" [Success] Schemas and SymPy physical laws formally verified on attempt {attempt}.")
print(f" SymPy Invariant Check: {physics_report}")
self.logger.log_generated_schemas(code_str, attempts=attempt, physics_report=physics_report)
return compiled_bundle
except (SecurityError, SchemaCompilationError, PhysicsVerificationError) as err:
error_category = type(err).__name__
error_message = str(err)
print(f" [Warning] Attempt {attempt} failed ({error_category}): {error_message}")
self.logger.log_schema_repair_event(attempt, error_category, error_message)
if attempt == max_schema_attempts:
raise RuntimeError(
f"Metaprogrammer failed to synthesize valid schemas and SymPy equations after {max_schema_attempts} attempts. "
f"Last error: {error_message}"
) from err
current_prompt = f"""Your previous Pydantic/SymPy code submission failed validation.
Diagnostic Error ({error_category}):
{error_message}
Rejected Code:
```python
{code_str}
Correction Directives:
Fix the specific error identified above.
If this is a
PhysicsVerificationError, inspect the SymPy formulation inPhysicsVerificationSandbox.verify_physical_invariants(). Ensure all symbols have appropriate physical assumptions (positive=True,real=True), the governing equations correctly reflect established physics, and the assertion logic does not crash.Ensure all 5 required classes (
DomainProposalSchema,DomainDefectSchema,DomainAuditReportSchema,DomainFinalSpecificationSchema,PhysicsVerificationSandbox) are defined.Output the corrected code inside a single
python ...block. """raise RuntimeError("Unexpected exit from schema synthesis loop.")def run_debate(self, problem: str, constraints: str, max_iterations: int = 2) -> BaseModel: self.logger.log_problem(problem, constraints)
print("[1/5] Synthesizing schemas and verifying physical laws via SymPy...") bundle = self.synthesize_domain_schemas_with_physics(problem, constraints, max_schema_attempts=3) ProposalCls = bundle["DomainProposalSchema"] AuditCls = bundle["DomainAuditReportSchema"] FinalCls = bundle["DomainFinalSpecificationSchema"] print("\n[2/5] Generating initial candidate proposal against verified physics...") gen_sys_inst = ( "You are a Principal Systems Architect. Formulate an uncompromised, " "rigorous engineering proposal that strictly resolves the stated contradictions. " "Respect the governing equations and physical invariants. Quantify all parameters." ) gen_prompt = f"Problem:\n{problem}\n\nConstraints:\n{constraints}\n\nPropose a complete technical solution." proposal = self._call_structured(self.architect_model, gen_sys_inst, gen_prompt, ProposalCls) self.logger.log_initial_proposal(proposal) audit_history: List[BaseModel] = [] for round_idx in range(1, max_iterations + 1): print(f"\n[3/5] Executing adversarial audit round {round_idx} of {max_iterations}...") critic_sys_inst = ( "You are an Adversarial Peer Reviewer, Red-Teamer, and Physics Auditor. " "Scrutinize the proposal strictly against governing physical laws, second-order parasitic " "penalties, unstated assumptions, and fatigue/thermal/boundary failure modes. " "Flag ungrounded optimism or unphysical claims. Set has_critical_defects=True " "if any physical or functional constraint is violated." ) audit_prompt = ( f"Problem Statement:\n{problem}\n\n" f"Governing Constraints:\n{constraints}\n\n" f"Candidate Design Proposal:\n{proposal.model_dump_json(indent=2)}\n\n" f"Perform the adversarial audit." ) audit = self._call_structured(self.critic_model, critic_sys_inst, audit_prompt, AuditCls) audit_history.append(audit) self.logger.log_audit_round(round_idx, audit) audit_dict = audit.model_dump() defects = audit_dict.get("identified_defects", []) critical_count = sum(1 for d in defects if d.get("severity") in ("CRITICAL", "HIGH")) has_crit = audit_dict.get("has_critical_defects", False) print(f" Identified {len(defects)} defects ({critical_count} critical/high).") if not has_crit or round_idx == max_iterations: if not has_crit: print(" Passed audit: All critical trade-offs successfully resolved.") else: print(" Max iterations reached. Proceeding to final synthesis with documented residual risks.") break print(f"\n[4/5] Rerouting defects to Architect for iteration {round_idx + 1}...") rev_sys_inst = ( "You are the Principal Systems Architect addressing hostile audit findings. " "Directly remediate all flagged physical failure modes and parasitic trade-offs " "by adjusting dimensions, material specs, or operational parameters." ) defects_summary = "\n".join([ f"- [{d.get('severity')}] {d.get('category')}: {d.get('description')} -> Fix: {d.get('recommended_mitigation')}" for d in defects ]) rev_prompt = ( f"Problem & Constraints:\n{problem}\n{constraints}\n\n" f"Prior Proposal:\n{proposal.model_dump_json(indent=2)}\n\n" f"Auditor Defects to Resolve:\n{defects_summary}\n\n" f"Produce a revised architectural proposal." ) revised_proposal = self._call_structured(self.architect_model, rev_sys_inst, rev_prompt, ProposalCls) self.logger.log_iteration_diff(round_idx, proposal, revised_proposal) proposal = revised_proposal print("\n[5/5] Synthesizing final technical specification...") synth_sys_inst = ( "You are the Executive Technical Authority. Reconcile the design history, final iteration, " "and audit findings into an authoritative production specification. Convert unresolved edge-cases " "into documented residual operational risks." ) history_summaries = [f"Round {i}: {a.model_dump().get('critique_summary')}" for i, a in enumerate(audit_history, 1)] synth_prompt = ( f"Problem & Constraints:\n{problem}\n{constraints}\n\n" f"Vetted Proposal:\n{proposal.model_dump_json(indent=2)}\n\n" f"Audit Iteration History:\n" + "\n".join(history_summaries) + "\n\n" f"Produce the final reconciled technical specification." ) final_spec = self._call_structured(self.synthesizer_model, synth_sys_inst, synth_prompt, FinalCls) self.logger.log_final_specification(final_spec) self.logger.write_to_disk() print("Engine run complete.\n") return final_spec
---------------------------------------------------------------------------
Execution Example
---------------------------------------------------------------------------
if name == "main": problem = ( "Improve the aerodynamic efficiency of an endurance racing bicycle while preserving " "rider pedaling power output and biomechanical comfort over a 6-hour duration." )
constraints = (
"- Aerodynamic power equation: P_aero = 0.5 * rho * CdA * v^3 (rho = 1.225 kg/m^3, benchmark v = 12.5 m/s).\n"
"- Hip angle at Top Dead Center (TDC) must not be forced below 45 degrees, otherwise sustainable power degrades.\n"
"- Cervical spine extension must stay under 35 degrees to prevent neck extensor fatigue over 6 hours.\n"
"- Total bike system mass must remain <= 8.5 kg."
)
engine = DynamicMultiModelDebateEngine(log_file="bike_aerodynamics_with_sympy.md")
final_design = engine.run_debate(problem, constraints, max_iterations=2)
Want to inspect the generated SymPy equations directly in the generated Markdown audit log?
query="Show me how to format the SymPy symbolic expressions as LaTeX equations inside the generated Markdown audit report."
Next steps to deepen the symbolic physics integration:
Add numerical evaluation of the proposal's parameters against the SymPy equations
query="Update the Auditor stage to substitute the numerical parameters from DomainProposalSchema into the SymPy equations to check for numerical conservation violations.
Export verified SymPy equations to LaTeX and AST graph plots
query="Add a helper to export the verified SymPy expression tree into LaTeX and an ASCII/NetworkX AST graph representation.
From <https://gemini.google.com/app/bbbc93cb53407388> Google Gemini (3.8 Flash)
Comments
Post a Comment