Add SIA: Self-Improving AI with Harness & Weight Updates

SIA/.gitignore

@@ -0,0 +1,2 @@
+__pycache__/
+*.pyc

SIA/__init__.py

@@ -0,0 +1,44 @@
+"""SIA: Self-Improving AI with Harness & Weight Updates.
+
+A configurable loop in which a language-model agent (the Feedback-Agent)
+updates both the harness and the weights of a task-specific agent.
+
+Reference: Hebbar et al., 2026. arXiv:2605.27276
+
+Quick start
+-----------
+from SIA import SIA
+from SIA.tasks import LawBenchTask
+
+task = LawBenchTask()
+sia = SIA(
+    task_spec=task.task_spec,
+    dataset=task.sample_instances(),
+    verifier=task.verifier,
+    g_max=5,
+    reference_impls=task.reference_impl,
+)
+result = sia.run()
+print(f"Best mean reward: {result.best_mean_reward:.4f}")
+"""
+from .sia_loop import SIA, SIAResult, GenerationRecord
+from .meta_agent import MetaAgent
+from .feedback_agent import FeedbackAgent
+from .task_agent import TaskAgent
+from .trajectory import Trajectory, Step, ToolCall
+from .verifier import Verifier, ExactMatchVerifier, FunctionVerifier
+
+__all__ = [
+    "SIA",
+    "SIAResult",
+    "GenerationRecord",
+    "MetaAgent",
+    "FeedbackAgent",
+    "TaskAgent",
+    "Trajectory",
+    "Step",
+    "ToolCall",
+    "Verifier",
+    "ExactMatchVerifier",
+    "FunctionVerifier",
+]

SIA/feedback_agent.py

@@ -0,0 +1,172 @@
+"""Feedback-Agent F: analyses trajectory τg and selects the next action.
+
+Action choices (§5.1):
+  - harness_update: synthesise an improved scaffold Ag+1 (weights fixed)
+  - weight_update:  trigger an RL weight-update step (scaffold fixed)
+
+Uses Claude Sonnet 4.6 as the LLM backbone (§5.2).
+"""
+from __future__ import annotations
+
+import json
+import textwrap
+from dataclasses import dataclass
+from typing import Any
+
+import anthropic
+
+from .trajectory import Trajectory
+from .weight_updates import ALGORITHM_REGISTRY, WeightUpdateAlgorithm
+
+_FB_SYSTEM = textwrap.dedent("""
+    You are the Feedback-Agent in the SIA (Self-Improving AI) framework.
+
+    You receive:
+    - The current scaffold source code (Ag)
+    - The execution trajectory τg (structured log of prompts, responses, tool calls,
+      extracted answers, and per-instance rewards)
+    - Performance metrics Eg
+    - The original task specification U
+    - Sample task descriptions (to help avoid over-fitting fixes to a single instance)
+
+    You must decide one of two actions:
+      1. "harness_update" — rewrite the scaffold to fix systemic issues
+         (parsing bugs, missing tools, bad retry logic, poor prompting strategy).
+         Return a JSON object: {"action": "harness_update", "new_scaffold": "<full Python source>",
+                                "report": "<prose analysis of changes>"}
+      2. "weight_update"  — trigger RL training on the current rollouts when the harness
+         has plateaued and domain-specific model knowledge is the bottleneck.
+         Choose the most appropriate algorithm from:
+           ppo_gae       — dense step-level rewards, stability critical
+           grpo          — cheap rollouts, episode-end verifier
+           entropic      — sparse/right-skewed rewards
+           reinforce_kl  — dense reward, capability regression risk
+           best_of_n     — near-zero pass rate, cold-start needed
+           dpo           — ordinal ranking possible, no cardinal reward
+         Return a JSON object: {"action": "weight_update", "algorithm": "<name>",
+                                "report": "<rationale>"}
+
+    Output ONLY a valid JSON object — no prose, no markdown fences.
+""").strip()
+
+_FB_USER_TMPL = textwrap.dedent("""
+    ## Current scaffold (Ag)
+    ```python
+    {scaffold}
+    ```
+
+    ## Performance metrics (Eg)
+    {metrics}
+
+    ## Execution trajectory summary (τg — last {n_examples} instances)
+    {trajectory_summary}
+
+    ## Task specification (U)
+    {task_spec}
+
+    ## Sample task descriptions (for regularisation)
+    {sample_descriptions}
+
+    Select the next action.
+""").strip()
+
+
+@dataclass
+class FeedbackDecision:
+    action: str  # "harness_update" | "weight_update"
+    new_scaffold: str | None = None
+    algorithm: str | None = None
+    report: str = ""
+    raw: str = ""
+
+
+class FeedbackAgent:
+    """Implements the Feedback-Agent decision loop."""
+
+    def __init__(
+        self,
+        model: str = "claude-sonnet-4-6",
+        max_tokens: int = 8192,
+        trajectory_examples: int = 5,
+    ):
+        self.client = anthropic.Anthropic()
+        self.model = model
+        self.max_tokens = max_tokens
+        self.trajectory_examples = trajectory_examples
+
+    def decide(
+        self,
+        scaffold: str,
+        trajectory: Trajectory,
+        task_spec: str,
+        sample_descriptions: list[str] | None = None,
+    ) -> FeedbackDecision:
+        """Return a FeedbackDecision given the current generation's artefacts."""
+        metrics_text = json.dumps(trajectory.metrics, indent=2)
+        trajectory_summary = _summarise_trajectory(trajectory, self.trajectory_examples)
+        samples_text = "\n".join(sample_descriptions or []) or "(none)"
+
+        user_msg = _FB_USER_TMPL.format(
+            scaffold=scaffold,
+            metrics=metrics_text,
+            n_examples=self.trajectory_examples,
+            trajectory_summary=trajectory_summary,
+            task_spec=task_spec,
+            sample_descriptions=samples_text,
+        )
+
+        response = self.client.messages.create(
+            model=self.model,
+            max_tokens=self.max_tokens,
+            system=_FB_SYSTEM,
+            messages=[{"role": "user", "content": user_msg}],
+        )
+        raw = response.content[0].text.strip()
+        return _parse_decision(raw)
+
+    def get_algorithm(self, name: str) -> WeightUpdateAlgorithm:
+        cls = ALGORITHM_REGISTRY.get(name)
+        if cls is None:
+            raise ValueError(f"Unknown weight-update algorithm: {name!r}. "
+                             f"Valid options: {list(ALGORITHM_REGISTRY)}")
+        return cls()
+
+
+def _summarise_trajectory(traj: Trajectory, n: int) -> str:
+    lines = []
+    for step in traj.steps[-n:]:
+        lines.append(
+            f"instance={step.instance_id!r} "
+            f"reward={step.reward} "
+            f"answer={step.extracted_answer!r}\n"
+            f"  response_snippet={step.response[:200]!r}"
+        )
+        if step.tool_calls:
+            for tc in step.tool_calls[:2]:
+                lines.append(f"  tool={tc.tool!r} error={tc.error!r}")
+    return "\n".join(lines) if lines else "(no steps)"
+
+
+def _parse_decision(raw: str) -> FeedbackDecision:
+    try:
+        data: dict[str, Any] = json.loads(raw)
+        action = data.get("action", "")
+        report = data.get("report", "")
+        if action == "harness_update":
+            return FeedbackDecision(
+                action="harness_update",
+                new_scaffold=data.get("new_scaffold", ""),
+                report=report,
+                raw=raw,
+            )
+        elif action == "weight_update":
+            return FeedbackDecision(
+                action="weight_update",
+                algorithm=data.get("algorithm", "grpo"),
+                report=report,
+                raw=raw,
+            )
+        else:
+            return FeedbackDecision(action="harness_update", report=f"Unparseable action: {action}", raw=raw)
+    except json.JSONDecodeError:
+        return FeedbackDecision(action="harness_update", report="JSON parse error", raw=raw)

SIA/meta_agent.py

@@ -0,0 +1,76 @@
+"""Meta-Agent M: generates the initial task-specific scaffold A1 from U and R.
+
+Uses Claude Sonnet 4.6 as the LLM backbone (§5.2).
+"""
+from __future__ import annotations
+
+import textwrap
+from typing import Any
+
+import anthropic
+
+_META_SYSTEM = textwrap.dedent("""
+    You are the Meta-Agent in the SIA (Self-Improving AI) framework.
+    Your job is to generate a complete, runnable Python scaffold for a task-specific agent.
+
+    The scaffold must:
+    1. Accept a task instance and produce an answer.
+    2. Include a system prompt, tool-dispatch logic, and answer extraction.
+    3. Expose a `run(instance: dict) -> dict` function that returns
+       {"answer": <prediction>, "trajectory_step": <Step dict>}.
+    4. Be self-contained (all imports at the top, no undefined references).
+
+    Output ONLY valid Python source code — no prose, no markdown fences.
+""").strip()
+
+_META_USER_TMPL = textwrap.dedent("""
+    Task specification:
+    {task_spec}
+
+    Reference implementations (if any):
+    {reference_impls}
+
+    Diverse sample instances to avoid overfitting the scaffold to a single case:
+    {sample_instances}
+
+    Generate the initial scaffold A1.
+""").strip()
+
+
+class MetaAgent:
+    """Generates A1 = M(U, R)."""
+
+    def __init__(self, model: str = "claude-sonnet-4-6", max_tokens: int = 8192):
+        self.client = anthropic.Anthropic()
+        self.model = model
+        self.max_tokens = max_tokens
+
+    def generate_scaffold(
+        self,
+        task_spec: str,
+        reference_impls: str = "",
+        sample_instances: list[dict[str, Any]] | None = None,
+    ) -> str:
+        """Return the source code of the initial scaffold A1."""
+        samples_text = _format_samples(sample_instances or [])
+        user_msg = _META_USER_TMPL.format(
+            task_spec=task_spec,
+            reference_impls=reference_impls or "(none provided)",
+            sample_instances=samples_text,
+        )
+        response = self.client.messages.create(
+            model=self.model,
+            max_tokens=self.max_tokens,
+            system=_META_SYSTEM,
+            messages=[{"role": "user", "content": user_msg}],
+        )
+        return response.content[0].text.strip()
+
+
+def _format_samples(samples: list[dict[str, Any]]) -> str:
+    if not samples:
+        return "(none provided)"
+    lines = []
+    for i, s in enumerate(samples[:5], 1):
+        lines.append(f"Sample {i}: {s}")
+    return "\n".join(lines)

SIA/requirements.txt

@@ -0,0 +1,3 @@
+anthropic>=0.40.0
+numpy>=1.24.0
+scikit-learn>=1.3.0