j4flmao/agent-skills/core-architectures

Variant: {react|plan-and-solve|plan-execute-replan|reflexion}
State: {fields}
Step: assemble_context -> decide -> validate -> execute -> observe -> update
Stop when: {success test} OR steps>={n} OR no_progress>={k} OR budget>={usd}
Reflection: {trigger, input signal, where lessons are stored}

flowchart TD
    Start([Goal]) --> Ctx[Assemble context: goal, plan, recent observations]
    Ctx --> Decide[Model proposes action or final answer]
    Decide --> Check{Final answer?}
    Check -- yes --> Verify{Success predicate holds?}
    Verify -- yes --> Done([Return])
    Verify -- no --> Reflect[Record failure reason] --> Ctx
    Check -- no --> Validate{Schema + policy valid?}
    Validate -- no --> Ctx
    Validate -- yes --> Exec[Execute tool] --> Obs[Append observation, update state]
    Obs --> Guard{Caps or no-progress hit?}
    Guard -- yes --> Abort([Abort with partial result])
    Guard -- no --> Ctx

from __future__ import annotations
import hashlib
import json
from dataclasses import dataclass, field
from typing import Any, Callable, Protocol
class Policy(Protocol):
    def decide(self, context: str) -> dict[str, Any]: ...  # {"type": "tool"|"final", ...}
@dataclass
class LoopState:
    goal: str
    plan: list[str] = field(default_factory=list)
    observations: list[str] = field(default_factory=list)
    lessons: list[str] = field(default_factory=list)
    seen_calls: dict[str, int] = field(default_factory=dict)
    steps: int = 0
    steps_without_progress: int = 0
class LoopAborted(RuntimeError):
    pass
def run_loop(
    state: LoopState,
    policy: Policy,
    tools: dict[str, Callable[..., str]],
    is_success: Callable[[str, LoopState], bool],
    max_steps: int = 15,
    max_no_progress: int = 3,
) -> str:
    while state.steps < max_steps:
        state.steps += 1
        context = "\n".join([
            f"GOAL: {state.goal}",
            "PLAN:\n" + "\n".join(state.plan),
            "LESSONS:\n" + "\n".join(state.lessons[-5:]),
            "RECENT OBSERVATIONS:\n" + "\n".join(state.observations[-6:]),
        ])
        action = policy.decide(context)
        if action["type"] == "final":
            if is_success(action["answer"], state):
                return action["answer"]
            state.lessons.append(f"Answer rejected by success check at step {state.steps}")
            continue
        name, args = action["tool"], action.get("args", {})
        if name not in tools:
            state.observations.append(f"ERROR: unknown tool '{name}'. Valid: {sorted(tools)}")
            continue
        key = hashlib.sha256(json.dumps([name, args], sort_keys=True).encode()).hexdigest()
        state.seen_calls[key] = state.seen_calls.get(key, 0) + 1
        if state.seen_calls[key] > 2:
            state.observations.append(f"NOTE: {name} called 3x with identical args; change approach.")
            state.steps_without_progress += 1
        else:
            try:
                result = tools[name](**args)
                state.observations.append(f"{name} -> {result[:2000]}")
                state.steps_without_progress = 0
            except Exception as exc:
                state.observations.append(f"{name} FAILED: {type(exc).__name__}: {exc}")
                state.steps_without_progress += 1
        if state.steps_without_progress >= max_no_progress:
            raise LoopAborted(f"no progress for {max_no_progress} steps; last: {state.observations[-1]}")
    raise LoopAborted(f"step cap {max_steps} reached for goal: {state.goal[:80]}")