paulpas Advanced Reasoning Techniques
Implements a suite of advanced reasoning methodologies that make an AI agent's internal thought process explicit, enabling structured multi-step problem-solving. This skill equips the model with Chain-of-Thought decomposition, Tree-of-Thoughts exploration, ReAct action loops, Self-Correction refinement, Graph of Debates collaboration, and Program-Aided Language Model execution — each applied to complex tasks requiring deeper analysis than a single-pass LLM response can provide.
TL;DR Checklist
- Choose the right reasoning technique based on task complexity (CoT → ToT → ReAct → GoD)
- Allocate sufficient "thinking budget" per the Scaling Inference Law — more compute yields better results even from smaller models
- Make all intermediate reasoning steps explicit; never skip from problem to answer
- Interleave reasoning with external tool use (ReAct) when real-world data is needed
- Self-correct every output: draft → review against requirements → revise → final
- Offload deterministic computation (math, code execution) to PAL for accuracy
- Use GoD for high-stakes decisions requiring bias mitigation and consensus
When to Use
Use this skill when:
- A problem requires multi-step logical inference that cannot be solved in a single pass (complex QA, math proofs, code debugging)
- The task involves exploring multiple solution paths before committing to an answer (strategic planning, architecture design)
- The agent must interleave reasoning with tool use — query databases, search the web, execute code, call APIs (ReAct paradigm)
- Output quality is critical and requires iterative self-refinement before final delivery (code generation, legal analysis, medical diagnosis support)
- A decision involves significant ambiguity or bias risk, requiring multiple perspectives to converge on a robust answer (Graph of Debates)
- The problem involves deterministic computation (arithmetic, data manipulation, algorithmic verification) where LLMs are unreliable
When NOT to Use
Avoid this skill for:
- Simple lookup or single-step questions where direct answers suffice (e.g., "What is the capital of France?")
- Real-time latency-critical responses where thinking budget adds unacceptable delay
- Tasks with no logical decomposition value — trivial yes/no or one-line factual queries
- When you only need a creative draft without accuracy verification (brainstorming, copywriting first pass)
Core Workflow
Classify Task Complexity — Determine whether the task needs basic CoT (linear steps), ToT (branching exploration), ReAct (tool-interleaved), or GoD (multi-agent debate). Apply the Scaling Inference Law: a smaller model with more thinking time often outperforms a large model with minimal reasoning. Checkpoint: Confirm the chosen technique matches task complexity before proceeding.
Decompose Into Reasoning Steps — Break the problem into a sequence of explicit intermediate steps (CoT) or generate multiple candidate reasoning paths at each branching point (ToT). Document each step's purpose and expected output. Checkpoint: Every decomposition step should be independently verifiable against the original requirements.
Execute Reasoning With Appropriate Depth — Run the selected technique: produce a thought-action-observation loop for ReAct, explore top-k branches with evaluation scoring for ToT, or generate candidate arguments for GoD nodes. Ensure each reasoning pass produces observable, checkable intermediate results. Checkpoint: All intermediate outputs are captured and can be audited.
Self-Correction Pass — Review every generated answer against the original requirements: accuracy (factual correctness), completeness (all aspects addressed), clarity (readable and concise), and tone alignment. Identify discrepancies, propose specific improvements, and generate a revised version. Checkpoint: The revised content addresses all identified weaknesses from step 3.
Offload Deterministic Computation — For any arithmetic, code execution, or data transformation within the reasoning chain, delegate to PAL: generate executable Python, run it in a sandboxed environment, and use the returned results in subsequent steps. Checkpoint: Code execution output matches the expected computation result; validate before incorporating into final answer.
Synthesize Final Output — Combine all validated intermediate results into a structured final answer with citations where applicable. For GoD, identify the most robust argument cluster based on verifiable knowledge or consensus strength. Checkpoint: Final output is coherent, complete, and traceable to explicit reasoning steps.
Implementation Patterns / Reference Guide
Pattern 1: Chain-of-Thought (CoT) Decomposition
Chain-of-Thought prompting guides the model through a step-by-step internal monologue before producing an answer. This transforms a single difficult problem into a sequence of simpler, verifiable sub-steps. Implement CoT by defining a persona, specifying the number and nature of reasoning steps, and capturing both the thought process and final answer.
from typing import Any
def build_cot_prompt(query: str, persona: str, step_count: int = 5) -> str:
"""Build a Chain-of-Thought prompt with structured reasoning steps.
Args:
query: The user's question or problem to solve.
persona: The role/identity the model should adopt.
step_count: Number of explicit reasoning steps (default 5).
Returns:
A formatted prompt that enforces step-by-step reasoning.
"""
# Define step templates based on common reasoning patterns
step_templates = [
"Analyze the Query", # Understand requirements
"Formulate Approach", # Plan the solution strategy
"Execute Reasoning Step", # Perform intermediate work
"Validate Intermediate Result",# Check correctness so far
"Synthesize Final Answer", # Produce polished output
]
steps_text = ""
for i, step in enumerate(step_templates[:step_count], 1):
steps_text += f"{i}. **{step}:** Describe what you should do at this stage.\n"
prompt = f"""You are an {persona}. Your goal is to answer the user's question
comprehensively and accurately by thinking step-by-step.
Here's the process you must follow:
{steps_text}
**User Query:** "{query}"
**Agent's Thought Process (Internal CoT Output):**
"""
return prompt
def execute_cot_reasoning(
thought_process: list[str],
query: str,
) -> dict[str, Any]:
"""Validate a Chain-of-Thought reasoning trace.
Args:
thought_process: List of intermediate reasoning steps produced by the model.
query: The original user query for reference.
Returns:
Dict with 'valid' boolean, 'step_count', and 'gaps' found during review.
"""
gaps: list[str] = []
# Check 1: At least one thought step exists
if len(thought_process) < 2:
gaps.append("Insufficient reasoning depth — expected at least 2 steps")
# Check 2: Each step references the query or previous step
for idx, step in enumerate(thought_process):
if idx == 0 and not any(kw in step.lower() for kw in ["query", "question", "user"]):
gaps.append(f"Step 1 does not reference the original query")
# Check 3: Final step should lead to a conclusion
last_step = thought_process[-1].lower()
if not any(kw in last_step for kw in ["conclusion", "final", "answer", "therefore", "result"]):
gaps.append("Final step does not produce a clear conclusion")
return {
"valid": len(gaps) == 0,
"step_count": len(thought_process),
"gaps": gaps,
}
BAD — Direct answer without reasoning trace:
Question: What is 15% tip on $84.50?
Answer: $12.68
❌ No reasoning visible. Cannot verify correctness. Single-pass hallucination risk high.
GOOD — CoT with explicit intermediate computation:
Question: What is 15% tip on $84.50?
**Agent's Thought Process:**
Thought 1 (Analyze): Need to calculate 15% of $84.50 for the tip amount.
Thought 2 (Plan): Convert percentage to decimal (0.15), multiply by base amount.
Thought 3 (Compute): 84.50 * 0.15 = 12.675
Thought 4 (Round): Round to nearest cent: $12.68
Thought 5 (Validate): Check — 10% of 84.50 is 8.45, 5% is 4.225, sum = 12.675 → rounds to 12.68. Correct.
**Final Answer:** The tip is $12.68
✅ Each step verifiable. Rounding logic explicit. Cross-validation included.
Pattern 2: Tree-of-Thoughts (ToT) Exploration
Tree-of-Thoughts extends CoT by branching at each reasoning step into multiple candidate thoughts, evaluating each branch before committing. This enables backtracking and exploration of alternative strategies — critical for tasks where the first obvious path may be suboptimal.
from typing import Any
class ThoughtNode:
"""A single node in a Tree-of-Thoughts reasoning tree.
Attributes:
thought: The reasoning content at this node.
score: Evaluation score (0.0-1.0) of this thought's promise.
children: List of child nodes generated from this thought.
parent: Reference to the parent ThoughtNode, or None for root.
"""
def __init__(self, thought: str, parent: "ThoughtNode | None" = None) -> None:
self.thought: str = thought
self.score: float = 0.0
self.children: list["ThoughtNode"] = []
self.parent: ThoughtNode | None = parent
def add_child(self, child: "ThoughtNode") -> None:
"""Add a child node and link parent reference."""
child.parent = self
self.children.append(child)
def to_path(self) -> list[str]:
"""Trace this node's ancestry back to root as a complete reasoning path."""
path: list[str] = []
node: ThoughtNode | None = self
while node is not None:
path.append(node.thought)
node = node.parent
return list(reversed(path))
class TreeOfThoughts:
"""Tree-of-Thoughts reasoning engine for exploring multiple solution paths.
Implements breadth-first exploration with evaluation and pruning at each depth level.
"""
def __init__(
self,
problem: str,
branches_per_step: int = 3,
max_depth: int = 4,
) -> None:
self.problem: str = problem
self.branches_per_step: int = branches_per_step
self.max_depth: int = max_depth
self.root: ThoughtNode | None = None
def generate_candidates(
self,
parent_node: ThoughtNode,
depth: int,
) -> list[ThoughtNode]:
"""Generate candidate thoughts branching from a parent node.
In production, this would call an LLM with the problem context plus
the parent's thought. Here we demonstrate the structure.
Args:
parent_node: The ThoughtNode to branch from.
depth: Current depth in the tree.
Returns:
List of new ThoughtNode candidates.
"""
if depth >= self.max_depth:
return []
# Production: call LLM with prompt like:
# "Given problem '{self.problem}' and parent thought: {parent_node.thought}
# Generate {self.branches_per_step} candidate next thoughts."
candidates: list[ThoughtNode] = []
for i in range(self.branches_per_step):
child = ThoughtNode(
thought=f"[Branch {i+1}] Consider an alternative approach...",
parent=parent_node,
)
candidates.append(child)
return candidates
def evaluate_thought(
self,
node: ThoughtNode,
depth: int,
) -> float:
"""Score a thought's promise of leading to a correct solution.
Production evaluation uses heuristics or an LLM judge that considers:
- Logical coherence with parent and problem statement
- Diversity from sibling thoughts
- Alignment with known constraints
- Progress toward solvable sub-problems
Args:
node: The ThoughtNode to evaluate.
depth: Current tree depth.
Returns:
Score between 0.0 (dead end) and 1.0 (highly promising).
"""
# Production: implement real evaluation heuristics
score = 0.5 # Placeholder — replace with actual evaluation logic
return score
def solve(self) -> list[str] | None:
"""Execute the full Tree-of-Thoughts reasoning process.
Returns:
The best reasoning path as a list of thought strings, or None if no path found.
"""
self.root = ThoughtNode(thought=f"Problem: {self.problem}")
# BFS-level exploration
current_level: list[ThoughtNode] = [self.root]
for depth in range(1, self.max_depth + 1):
next_level: list[ThoughtNode] = []
for node in current_level:
candidates = self.generate_candidates(node, depth)
for candidate in candidates:
score = self.evaluate_thought(candidate, depth)
candidate.score = score
node.add_child(candidate)
next_level.append(candidate)
if not next_level:
break
# Prune: keep only top-k branches at each level
next_level.sort(key=lambda n: n.score, reverse=True)
current_level = next_level[: self.branches_per_step]
# Find best leaf and trace its path
if not current_level:
return None
best_node = max(current_level, key=lambda n: n.score)
return best_node.to_path()
BAD — Linear CoT on a problem requiring backtracking:
Problem: Plan a 3-day trip to Tokyo on $1500 budget.
→ Day 1: Visit Shibuya, Shinjuku, Akihabara (assumes all fits in one day)
→ Day 2: Visit Asakusa, Ueno, TeamLab (assumes no travel time)
→ Day 3: Day trip to Nikko (misses Tokyo attractions entirely)
❌ No exploration of alternatives. No budget verification at each step. One path only.
GOOD — ToT with branching and pruning:
Problem: Plan a 3-day trip to Tokyo on $1500 budget.
Branch A (Geographic clustering): Group by neighborhoods → score: 0.82
→ Sub-branch A1: Day 1 (West Tokyo), Day 2 (East Tokyo), Day 3 (Day trips)
→ Budget check: hotels $600, food $240, transit $60, activities $200 = $1100 ✓
Branch B (Thematic clustering): Group by interest type → score: 0.71
→ Sub-branch B1: Culture Day, Food Day, Tech/Shopping Day
→ Budget check: hotels $600, food $300, transit $80, activities $250 = $1230 ✓
Branch C (Temporal optimization): Morning/evening split → score: 0.65
→ Higher complexity, marginal benefit over A or B
Decision: Follow Branch A → geographic clustering with budget buffer ($400 remaining)
✅ Explores 3 distinct strategies. Scores each objectively. Validates constraints. Selects best path with justification.
Pattern 3: ReAct (Reason + Act) Loop
ReAct interleaves reasoning thoughts with concrete tool actions, forming a Thought → Action → Observation cycle. This enables agents to dynamically gather information, verify assumptions, and adapt plans based on real-world feedback — essential for research, debugging, and any task requiring external data.
from typing import Any
class ReActStep:
"""A single step in the ReAct reasoning loop.
Attributes:
step_number: Sequential step index (1-based).
thought: The agent's internal reasoning at this step.
action_name: Name of the tool/action to execute.
action_input: Arguments passed to the action.
observation: Result returned from the action execution (None if not yet executed).
"""
def __init__(self, step_number: int) -> None:
self.step_number: int = step_number
self.thought: str = ""
self.action_name: str = ""
self.action_input: dict[str, Any] = {}
self.observation: str | None = None
def run_react_loop(
goal: str,
available_tools: dict[str, callable],
max_steps: int = 10,
) -> dict[str, Any]:
"""Execute a ReAct reasoning loop with tool-interleaved action.
Args:
goal: The task the agent must accomplish.
available_tools: Mapping of tool names to executable functions.
max_steps: Maximum number of Thought-Action-Observation cycles.
Returns:
Dict with 'final_answer', 'steps' (list of ReActStep), and 'terminated_early' bool.
"""
steps: list[ReActStep] = []
current_step_idx: int = 1
terminated_early: bool = False
while current_step_idx <= max_steps:
step = ReActStep(step_number=current_step_idx)
# --- THOUGHT Phase: Reason about what to do next ---
all_obs = [s.observation for s in steps if s.observation is not None]
context = f"Goal: {goal}\nPrevious observations:\n" + "\n".join(all_obs)
step.thought = _generate_thought(context, available_tools, current_step_idx)
# Check if the thought indicates a "finish" action
if _is_finish_thought(step.thought):
step.action_name = "finish"
step.action_input = {"answer": _extract_final_answer(step.thought)}
step.observation = None
steps.append(step)
terminated_early = True
break
# --- ACTION Phase: Select and execute a tool ---
action_name, action_input = _select_action(
step.thought, available_tools, current_step_idx
)
step.action_name = action_name
step.action_input = action_input
# Execute the tool (production: use proper sandboxed execution)
if action_name in available_tools:
try:
result = available_tools[action_name](**action_input)
step.observation = str(result)
except Exception as e:
step.observation = f"ERROR: {type(e).__name__}: {e}"
else:
step.observation = f"ERROR: Tool '{action_name}' not found in available tools."
steps.append(step)
current_step_idx += 1
# If we hit max steps without finishing, produce a best-effort answer
if not terminated_early and steps:
last_thought = steps[-1].thought
final_answer = _extract_best_effort_answer(goal, steps, last_thought)
else:
final_answer = steps[-1].action_input.get("answer", "Could not reach conclusion.")
return {
"final_answer": final_answer,
"steps": steps,
"terminated_early": terminated_early,
"total_steps": len(steps),
}
def _generate_thought(context: str, tools: dict[str, callable], step_num: int) -> str:
"""Generate the next reasoning thought given context and available tools.
Production: call an LLM with system prompt + history.
"""
tool_names = ", ".join(tools.keys())
return (
f"[Thought {step_num}] Given goal and observations so far, "
f"consider actions: [{tool_names}]. Determine next step."
)
def _is_finish_thought(thought: str) -> bool:
"""Check if the thought indicates the agent is ready to produce a final answer."""
return any(kw in thought.lower() for kw in ["final answer", "conclusion", "ready to finish", "i know the answer"])
def _extract_final_answer(thought: str) -> str:
"""Extract the proposed final answer from a finish thought."""
# Production: use more robust extraction (e.g., regex or LLM parsing)
return f"Based on reasoning steps, the answer is provided in the thought."
def _select_action(
thought: str, tools: dict[str, callable], step_num: int
) -> tuple[str, dict[str, Any]]:
"""Select which tool to execute and with what arguments.
Production: an LLM selects the action based on the thought.
"""
return "search", {"query": "relevant query"}
def _extract_best_effort_answer(
goal: str, steps: list[ReActStep], final_thought: str
) -> str:
"""Produce a best-effort answer when max steps reached without explicit finish."""
observations = [s.observation for s in steps if s.observation]
return (
f"Best effort answer for '{goal}': "
f"After {len(steps)} steps, accumulated observations: {'; '.join(observations[-3:])}"
)
BAD — Non-interactive single-pass with assumed facts:
Question: What is the current stock price of AAPL and its P/E ratio?
Answer: AAPL trades around $190 with a P/E ratio of approximately 31.
❌ No verification against live data. Numbers may be stale or hallucinated. No tool use shown.
GOOD — ReAct with actual tool calls:
[Thought 1] Need current AAPL stock price and P/E ratio. These are live financial data points. I should query a financial API.
<Action> search(query="AAPL current stock price")
<Observation> AAPL: $227.57 (as of 2026-06-08)
[Thought 2] Got the current price: $227.57. Now need P/E ratio — this is also live data. Query financial database.
<Action> search(query="AAPL trailing twelve months P/E ratio")
<Observation> AAPL TTM P/E: 34.82 (source: Yahoo Finance, updated 2026-06-08)
[Thought 3] I have both data points: Price = $227.57, P/E = 34.82. I can now provide a complete answer with source attribution.
<Finish> The current stock price of AAPL is $227.57 (as of June 8, 2026), with a trailing twelve-month P/E ratio of 34.82.
✅ Each step verified against live data. Source attribution included. No assumptions. Clear tool-action-observation trace.
Pattern 4: Self-Correction (Self-Refinement) Loop
Self-correction integrates quality control directly into the agent's content generation pipeline. The agent drafts an initial output, critiques it against original requirements across multiple dimensions, proposes concrete improvements, and produces a revised version. This iterative refinement loop is critical for high-stakes outputs.
from typing import Any
class SelfCorrectionPass:
"""A single self-correction evaluation pass over generated content.
Attributes:
original_prompt: The initial user requirements.
draft_content: The content being evaluated.
discrepancies: List of identified issues with severity and type.
improvements: List of concrete proposed fixes.
"""
def __init__(self, original_prompt: str, draft_content: str) -> None:
self.original_prompt: str = original_prompt
self.draft_content: str = draft_content
self.discrepancies: list[dict[str, str]] = []
self.improvements: list[dict[str, str]] = []
def evaluate(
self,
dimensions: list[str] | None = None,
) -> list[dict[str, str]]:
"""Evaluate content across multiple quality dimensions.
Args:
dimensions: Quality criteria to check (default: all 6 dimensions).
Returns:
List of discrepancy dicts with 'dimension', 'issue', and 'severity' keys.
"""
if dimensions is None:
dimensions = [
"accuracy", # Factual correctness
"completeness", # All requirements addressed
"clarity", # Readable and unambiguous
"tone", # Matches desired style
"engagement", # Captures attention
"conciseness", # No unnecessary verbosity
]
self.discrepancies = []
for dim in dimensions:
issues = self._check_dimension(dim)
self.discrepancies.extend(issues)
return self.discrepancies
def propose_improvements(self) -> list[dict[str, str]]:
"""Generate specific improvement proposals based on identified discrepancies.
Returns:
List of improvement dicts with 'dimension', 'issue', and 'action' keys.
"""
self.improvements = []
for disc in self.discrepancies:
improvement = {
"dimension": disc["dimension"],
"issue": disc["issue"],
"action": self._generate_fix_action(disc),
}
self.improvements.append(improvement)
return self.improvements
def _check_dimension(self, dimension: str) -> list[dict[str, str]]:
"""Check content against a single quality dimension.
Production: call an LLM with the content + dimension-specific rubric.
"""
findings: list[dict[str, str]] = []
if dimension == "accuracy":
# Check for factual claims that should be verified
if any(kw in self.draft_content.lower() for kw in ["2024", "2025"]):
findings.append({
"dimension": "accuracy",
"issue": "Contains date-specific claims that may be outdated.",
"severity": "high",
})
elif dimension == "completeness":
# Check if key entities from the prompt are mentioned
prompt_entities = self.original_prompt.split()[:10]
missing = [e for e in prompt_entities if e.lower() not in self.draft_content.lower()]
if len(missing) > 3:
findings.append({
"dimension": "completeness",
"issue": f"May be missing key topics from original prompt: {', '.join(missing[:3])}",
"severity": "high",
})
elif dimension == "clarity":
if len(self.draft_content.split()) > 500 and "." not in self.draft_content[-100:]:
findings.append({
"dimension": "clarity",
"issue": "Content appears to lack a proper conclusion.",
"severity": "medium",
})
return findings
def _generate_fix_action(self, discrepancy: dict[str, str]) -> str:
"""Generate a concrete fix action for a given discrepancy.
Production: an LLM generates specific rewrite suggestions.
"""
return f"Revise content to address {discrepancy['issue'].lower()}."
def apply_self_correction(
original_prompt: str,
draft_content: str,
max_iterations: int = 3,
) -> dict[str, Any]:
"""Apply a self-correction loop: draft → evaluate → improve → revise.
Args:
original_prompt: The user's original requirements.
draft_content: The initial content draft to refine.
max_iterations: Maximum refinement cycles before accepting the result.
Returns:
Dict with 'revised_content', 'iterations', and 'final_discrepancies'.
"""
current_content = draft_content
all_discrepancies: list[dict[str, str]] = []
for iteration in range(max_iterations):
evaluator = SelfCorrectionPass(original_prompt, current_content)
discrepancies = evaluator.evaluate()
if not discrepancies:
# Content passes all checks — no further refinement needed
break
all_discrepancies = discrepancies
improvements = evaluator.propose_improvements()
# Production: LLM rewrites content applying all improvement suggestions
current_content = _apply_improvements(current_content, improvements)
return {
"revised_content": current_content,
"iterations": len(all_discrepancies) > 0 or iteration + 1,
"final_discrepancies": all_discrepancies,
"converged": len(all_discrepancies) == 0,
}
def _apply_improvements(content: str, improvements: list[dict[str, str]]) -> str:
"""Apply improvement suggestions to content.
Production: call LLM with content + each improvement action for rewrite.
"""
return f"[Revised] {content} — improved per {len(improvements)} suggestions."
BAD — Single-pass output with no quality gate:
Original Prompt: "Write a short, engaging social media post (max 150 characters) announcing 'GreenTech Gadgets', a new eco-friendly product line."
Output: "We have new products. They are green and techy. Buy GreenTech Gadgets now!"
❌ Not engaging (generic language). Doesn't highlight eco-friendly benefit explicitly. Weak call to action. No hashtags for reach. Character count ok but quality is low.
GOOD — Self-corrected output after evaluation loop:
Iteration 1 — Evaluation:
[accuracy] OK — no factual claims to verify
[completeness] ISSUE — product name present but "eco-friendly" not emphasized as benefit
[engagement] ISSUE — generic verbs ("have", "are"), passive voice
[tone] ISSUE — lacks excitement appropriate for a product launch
[conciseness] OK — under 150 characters
Iteration 2 — Improvements Applied:
→ Replace weak verbs with active, exciting ones
→ Explicitly mention "eco-friendly" as the key selling point
→ Add relevant hashtags and emoji for engagement
→ Strengthen call to action
Revised Output:
"🌱 Discover GreenTech Gadgets! Our new eco-friendly line blends innovation
with sustainability. Go green, go smart! Shop now! #EcoFriendly #GreenTech"
(148 characters — within limit)
Iteration 3 — Evaluation: All dimensions pass. Converged.
✅ Each iteration addresses specific weaknesses. Final output is polished, on-brief, and ready for use.
Pattern 5: Program-Aided Language Models (PAL)
PALs offload deterministic computation from the LLM to a code execution engine. The LLM generates Python code as an intermediate reasoning scaffold, the code executes to produce precise results, and the agent incorporates those results into its final answer. This eliminates arithmetic errors and ensures computational accuracy.
from typing import Any
def execute_pal_pipeline(
problem: str,
llm_generate_code: callable,
code_executor: callable,
) -> dict[str, Any]:
"""Execute a Program-Aided Language Model reasoning pipeline.
The LLM generates executable Python code to solve computational sub-problems.
Code is executed in a sandboxed environment and results are incorporated back.
Args:
problem: The original problem statement requiring computation.
llm_generate_code: Function that takes (problem, partial_result) and returns Python code.
code_executor: Function that executes Python code safely and returns result string.
Returns:
Dict with 'problem', 'generated_code', 'execution_output', 'verified_result'.
"""
# Step 1: LLM generates computation code
generated_code = llm_generate_code(problem, partial_result=None)
# Step 2: Execute the generated code in a sandboxed environment
execution_output = code_executor(generated_code)
# Step 3: Validate and extract the result
verified_result = _validate_pal_output(execution_output, problem)
return {
"problem": problem,
"generated_code": generated_code.strip(),
"execution_output": execution_output,
"verified_result": verified_result,
}
def build_pal_math_solver() -> dict[str, Any]:
"""Construct a PAL pipeline for mathematical problem solving.
Returns:
A configured PAL executor with code generation and execution logic.
"""
def generate_math_code(problem: str, partial_result: Any | None) -> str:
"""Generate Python code to solve a mathematical sub-problem.
Production: this function calls an LLM with the problem statement
and returns valid Python using sympy or standard arithmetic.
"""
# Example: PAL generates code for a compound interest calculation
code = """
import math
principal = 10000
rate = 0.07 # 7% annual interest
time_years = 5
compounding_periods = 12 # monthly
# Compound interest formula: A = P(1 + r/n)^(nt)
amount = principal * (1 + rate / compounding_periods) ** (compounding_periods * time_years)
interest_earned = amount - principal
print(f"Final amount: ${amount:.2f}")
print(f"Interest earned: ${interest_earned:.2f}")
"""
return code.strip()
def execute_code(code: str) -> str:
"""Execute Python code in a sandboxed environment.
Production: use RestrictedPython, Docker sandbox, or similar
to prevent arbitrary code execution.
"""
# Example output for the compound interest problem above:
return "Final amount: $14176.25\nInterest earned: $4176.25"
result = execute_pal_pipeline(
problem="Calculate the compound interest on $10,000 at 7% annual rate over 5 years, compounded monthly.",
llm_generate_code=generate_math_code,
code_executor=execute_code,
)
return result
def build_pal_code_debugger() -> dict[str, Any]:
"""Construct a PAL pipeline for programmatic code debugging.
Generates and runs test cases to identify bugs in provided code.
"""
def generate_test_code(buggy_code: str, problem_description: str) -> str:
"""Generate Python test harness to exercise buggy code.
Production: LLM generates tests based on the problem specification
and the buggy implementation's interface.
"""
return """
# Test cases for the sorting function under investigation
from typing import list
def run_tests(sort_func):
results = []
# Test 1: Normal case — mixed integers
test_input = [3, 1, 4, 1, 5, 9, 2, 6]
expected = sorted(test_input)
actual = sort_func(test_input.copy())
results.append({"test": "normal_case", "pass": actual == expected, "actual": actual})
# Test 2: Edge case — empty list
test_input = []
actual = sort_func(test_input.copy())
results.append({"test": "empty_list", "pass": actual == [], "actual": actual})
# Test 3: Edge case — single element
test_input = [42]
actual = sort_func(test_input.copy())
results.append({"test": "single_element", "pass": actual == [42], "actual": actual})
# Test 4: Already sorted list
test_input = [1, 2, 3, 4, 5]
actual = sort_func(test_input.copy())
results.append({"test": "already_sorted", "pass": actual == [1,2,3,4,5], "actual": actual})
# Test 5: All same elements
test_input = [7, 7, 7, 7]
actual = sort_func(test_input.copy())
results.append({"test": "all_same", "pass": actual == [7,7,7,7], "actual": actual})
for r in results:
status = "PASS" if r["pass"] else "FAIL"
print(f"[{status}] {r['test']}: expected={expected if r['test']=='normal_case' else 'N/A'}, got={r['actual']}")
# Run all tests
run_tests(sort_func)
"""
return code.strip()
# Execute and analyze results
test_output = """[PASS] normal_case: expected=[1, 1, 2, 3, 4, 5, 6, 9], got=[1, 1, 2, 3, 4, 5, 6, 9]
[PASS] empty_list: expected=N/A, got=[]
[FAIL] single_element: expected=N/A, got=[]
[PASS] already_sorted: expected=N/A, got=[1, 2, 3, 4, 5]
[PASS] all_same: expected=N/A, got=[7, 7, 7, 7]"""
# Analyze: Test 3 fails — single element list returns empty instead of [42]
# Bug likely in a condition checking `if len(arr) > 1` that skips single-element case
return {
"problem": "Find bugs in the provided sorting function using test-driven debugging",
"test_code": generate_test_code("# buggy sort implementation", ""),
"test_results": test_output,
"diagnosis": "Bug identified: Single-element list returns empty (Test 3 FAIL). Check guard clause that requires len(arr) > 1.",
}
def _validate_pal_output(
execution_output: str,
problem: str,
) -> str | None:
"""Validate that PAL code execution produced a reasonable result.
Args:
execution_output: Raw output from code execution.
problem: Original problem for context.
Returns:
Validated result string, or None if execution failed.
"""
if not execution_output or "Error" in execution_output or "Traceback" in execution_output:
return None
return execution_output.strip()
BAD — LLM does math directly (hallucination-prone):
Problem: A rectangle has width 17.35m and length 24.82m. Calculate the area and perimeter.
LLM Answer: Area = 430.71 square meters. Perimeter = 84.34 meters.
❌ LLMs are notoriously bad at arithmetic. Numbers may be plausible but wrong. No verification mechanism.
GOOD — PAL generates and executes code for precise computation:
Problem: A rectangle has width 17.35m and length 24.82m. Calculate the area and perimeter.
Generated Python Code:
width = 17.35
length = 24.82
area = width * length
perimeter = 2 * (width + length)
print(f"Area: {area:.2f} m²")
print(f"Perimeter: {perimeter:.2f} m")
Execution Output:
Area: 430.62 m²
Perimeter: 84.34 m
Verified Result: Area = 430.62 m², Perimeter = 84.34 m
✅ Deterministic computation via code execution. No arithmetic hallucination. Exact results with proper rounding. Code is auditable and reproducible.
Pattern 6: Graph of Debates (GoD) Framework
Graph of Debates structures multi-agent reasoning as a dynamic, non-linear network where arguments are nodes connected by "supports" or "refutes" edges. This moves beyond linear debate chains to a richer topology that can dynamically branch, converge, and identify the most robust argument cluster.
from typing import Any
class DebateNode:
"""A single argument node in a Graph of Debates.
Attributes:
content: The argument text (claim or counter-claim).
node_id: Unique identifier for this node.
claim_type: Whether this node presents a 'claim', 'support', or 'refutation'.
evidence: List of verifiable evidence items supporting this argument.
confidence: Confidence score based on evidence strength and consensus (0.0-1.0).
children: Nodes that support or refute this one.
"""
def __init__(self, node_id: int, content: str, claim_type: str = "claim") -> None:
self.node_id: int = node_id
self.content: str = content
self.claim_type: str = claim_type # 'claim', 'support', or 'refutation'
self.evidence: list[str] = []
self.confidence: float = 0.5
self.children: list["DebateNode"] = []
self.parent_ids: list[int] = []
def add_evidence(self, evidence_item: str) -> None:
"""Add a verifiable evidence item to this argument."""
self.evidence.append(evidence_item)
class DebateGraph:
"""Graph of Debates (GoD) framework for multi-agent collaborative reasoning.
Arguments form nodes; edges represent 'supports' or 'refutes' relationships.
A conclusion emerges from the most robust, well-supported argument cluster.
"""
def __init__(self) -> None:
self.nodes: dict[int, DebateNode] = {}
self.next_id: int = 0
self.support_edges: list[tuple[int, int]] = [] # (parent_id, child_id)
self.refutes_edges: list[tuple[int, int]] = []
def add_argument(
self,
content: str,
claim_type: str = "claim",
evidence: list[str] | None = None,
) -> int:
"""Add a new argument node to the debate graph.
Args:
content: The argument text.
claim_type: 'claim', 'support', or 'refutation'.
evidence: Optional verifiable evidence items.
Returns:
The node ID assigned to this argument.
"""
node = DebateNode(self.next_id, content, claim_type)
if evidence:
node.evidence = evidence.copy()
self.nodes[self.next_id] = node
self.next_id += 1
return self.next_id - 1
def link_support(self, supporter_id: int, supported_id: int) -> None:
"""Add a 'supports' edge between two nodes."""
self.support_edges.append((supporter_id, supported_id))
if supported_id in self.nodes:
self.nodes[supported_id].childre
…(truncated)